feat!: add support for sharded pubsub topics & remove support for named pubsub topics

packages/core/src/lib/base_protocol.ts

@@ -2,9 +2,16 @@ import type { Libp2p } from "@libp2p/interface";
 import type { Stream } from "@libp2p/interface/connection";
 import type { PeerId } from "@libp2p/interface/peer-id";
 import { Peer, PeerStore } from "@libp2p/interface/peer-store";
-import type { IBaseProtocol, Libp2pComponents } from "@waku/interfaces";
+import type {
+  IBaseProtocol,
+  Libp2pComponents,
+  PubsubTopic,
+  ShardInfo
+} from "@waku/interfaces";
+import { shardInfoToPubsubTopics } from "@waku/utils";
 import { getPeersForProtocol, selectPeerForProtocol } from "@waku/utils/libp2p";
 
+import { DefaultPubsubTopic } from "./constants.js";
 import { filterPeers } from "./filterPeers.js";
 import { StreamManager } from "./stream_manager.js";
 
@@ -89,4 +96,10 @@ export class BaseProtocol implements IBaseProtocol {
     // Filter the peers based on the specified criteria
     return filterPeers(allPeersForProtocol, numPeers, maxBootstrapPeers);
   }
+
+  initializePubsubTopic(shardInfo?: ShardInfo): PubsubTopic[] {
+    return shardInfo
+      ? shardInfoToPubsubTopics(shardInfo)
+      : [DefaultPubsubTopic];
+  }
 }

packages/core/src/lib/filter/index.ts

@@ -14,12 +14,14 @@ import type {
   PeerIdStr,
   ProtocolCreateOptions,
   PubsubTopic,
+  SingleShardInfo,
   Unsubscribe
 } from "@waku/interfaces";
 import { WakuMessage } from "@waku/proto";
 import {
   ensurePubsubTopicIsConfigured,
   groupByContentTopic,
+  singleShardInfoToPubsubTopic,
   toAsyncIterator
 } from "@waku/utils";
 import { Logger } from "@waku/utils";
@@ -279,7 +281,7 @@ class Filter extends BaseProtocol implements IReceiver {
   constructor(libp2p: Libp2p, options?: ProtocolCreateOptions) {
     super(FilterCodecs.SUBSCRIBE, libp2p.components);
 
-    this.pubsubTopics = options?.pubsubTopics || [DefaultPubsubTopic];
+    this.pubsubTopics = this.initializePubsubTopic(options?.shardInfo);
 
     libp2p.handle(FilterCodecs.PUSH, this.onRequest.bind(this)).catch((e) => {
       log.error("Failed to register ", FilterCodecs.PUSH, e);
@@ -289,8 +291,12 @@ class Filter extends BaseProtocol implements IReceiver {
   }
 
   async createSubscription(
-    pubsubTopic: string = DefaultPubsubTopic
+    pubsubTopicShardInfo?: SingleShardInfo
   ): Promise<Subscription> {
+    const pubsubTopic = pubsubTopicShardInfo
+      ? singleShardInfoToPubsubTopic(pubsubTopicShardInfo)
+      : DefaultPubsubTopic;
+
     ensurePubsubTopicIsConfigured(pubsubTopic, this.pubsubTopics);
 
     //TODO: get a relevant peer for the topic/shard

packages/core/src/lib/keep_alive_manager.ts

@@ -2,7 +2,7 @@ import type { PeerId } from "@libp2p/interface/peer-id";
 import type { PeerStore } from "@libp2p/interface/peer-store";
 import type { IRelay, PeerIdStr } from "@waku/interfaces";
 import type { KeepAliveOptions } from "@waku/interfaces";
-import { Logger } from "@waku/utils";
+import { Logger, pubsubTopicToSingleShardInfo } from "@waku/utils";
 import { utf8ToBytes } from "@waku/utils/bytes";
 import type { PingService } from "libp2p/ping";
 
@@ -129,7 +129,7 @@ export class KeepAliveManager {
       if (!meshPeers.includes(peerIdStr)) continue;
 
       const encoder = createEncoder({
-        pubsubTopic: topic,
+        pubsubTopicShardInfo: pubsubTopicToSingleShardInfo(topic),
         contentTopic: RelayPingContentTopic,
         ephemeral: true
       });

packages/core/src/lib/light_push/index.ts

@@ -22,7 +22,6 @@ import { pipe } from "it-pipe";
 import { Uint8ArrayList } from "uint8arraylist";
 
 import { BaseProtocol } from "../base_protocol.js";
-import { DefaultPubsubTopic } from "../constants.js";
 
 import { PushRpc } from "./push_rpc.js";
 
@@ -50,7 +49,7 @@ class LightPush extends BaseProtocol implements ILightPush {
 
   constructor(libp2p: Libp2p, options?: ProtocolCreateOptions) {
     super(LightPushCodec, libp2p.components);
-    this.pubsubTopics = options?.pubsubTopics ?? [DefaultPubsubTopic];
+    this.pubsubTopics = this.initializePubsubTopic(options?.shardInfo);
   }
 
   private async preparePushMessage(

packages/core/src/lib/message/version_0.ts

@@ -7,10 +7,11 @@ import type {
   IMetaSetter,
   IProtoMessage,
   IRateLimitProof,
-  PubsubTopic
+  PubsubTopic,
+  SingleShardInfo
 } from "@waku/interfaces";
 import { proto_message as proto } from "@waku/proto";
-import { Logger } from "@waku/utils";
+import { Logger, singleShardInfoToPubsubTopic } from "@waku/utils";
 
 import { DefaultPubsubTopic } from "../constants.js";
 
@@ -119,12 +120,19 @@ export class Encoder implements IEncoder {
  * messages.
  */
 export function createEncoder({
-  pubsubTopic = DefaultPubsubTopic,
+  pubsubTopicShardInfo,
   contentTopic,
   ephemeral,
   metaSetter
 }: EncoderOptions): Encoder {
-  return new Encoder(contentTopic, ephemeral, pubsubTopic, metaSetter);
+  return new Encoder(
+    contentTopic,
+    ephemeral,
+    pubsubTopicShardInfo?.index
+      ? singleShardInfoToPubsubTopic(pubsubTopicShardInfo)
+      : DefaultPubsubTopic,
+    metaSetter
+  );
 }
 
 export class Decoder implements IDecoder<DecodedMessage> {
@@ -182,7 +190,12 @@ export class Decoder implements IDecoder<DecodedMessage> {
  */
 export function createDecoder(
   contentTopic: string,
-  pubsubTopic: PubsubTopic = DefaultPubsubTopic
+  pubsubTopicShardInfo?: SingleShardInfo
 ): Decoder {
-  return new Decoder(pubsubTopic, contentTopic);
+  return new Decoder(
+    pubsubTopicShardInfo?.index
+      ? singleShardInfoToPubsubTopic(pubsubTopicShardInfo)
+      : DefaultPubsubTopic,
+    contentTopic
+  );
 }

packages/core/src/lib/store/index.ts

@@ -19,7 +19,6 @@ import { pipe } from "it-pipe";
 import { Uint8ArrayList } from "uint8arraylist";
 
 import { BaseProtocol } from "../base_protocol.js";
-import { DefaultPubsubTopic } from "../constants.js";
 import { toProtoMessage } from "../to_proto_message.js";
 
 import { HistoryRpc, PageDirection, Params } from "./history_rpc.js";
@@ -80,7 +79,7 @@ class Store extends BaseProtocol implements IStore {
 
   constructor(libp2p: Libp2p, options?: ProtocolCreateOptions) {
     super(StoreCodec, libp2p.components);
-    this.pubsubTopics = options?.pubsubTopics ?? [DefaultPubsubTopic];
+    this.pubsubTopics = this.initializePubsubTopic(options?.shardInfo);
   }
 
   /**

packages/core/src/lib/waku.ts

@@ -8,12 +8,14 @@ import type {
   IStore,
   Libp2p,
   PubsubTopic,
+  ShardInfo,
   Waku
 } from "@waku/interfaces";
 import { Protocols } from "@waku/interfaces";
-import { Logger } from "@waku/utils";
+import { Logger, shardInfoToPubsubTopics } from "@waku/utils";
 
 import { ConnectionManager } from "./connection_manager.js";
+import { DefaultPubsubTopic } from "./constants.js";
 
 export const DefaultPingKeepAliveValueSecs = 5 * 60;
 export const DefaultRelayKeepAliveValueSecs = 5 * 60;
@@ -50,16 +52,23 @@ export class WakuNode implements Waku {
   public filter?: IFilter;
   public lightPush?: ILightPush;
   public connectionManager: ConnectionManager;
+  public readonly pubsubTopics: PubsubTopic[];
 
   constructor(
     options: WakuOptions,
-    public readonly pubsubTopics: PubsubTopic[],
     libp2p: Libp2p,
+    pubsubShardInfo?: ShardInfo,
     store?: (libp2p: Libp2p) => IStore,
     lightPush?: (libp2p: Libp2p) => ILightPush,
     filter?: (libp2p: Libp2p) => IFilter,
     relay?: (libp2p: Libp2p) => IRelay
   ) {
+    if (!pubsubShardInfo) {
+      this.pubsubTopics = [DefaultPubsubTopic];
+    } else {
+      this.pubsubTopics = shardInfoToPubsubTopics(pubsubShardInfo);
+    }
+
     this.libp2p = libp2p;
 
     if (store) {
@@ -88,7 +97,7 @@ export class WakuNode implements Waku {
       peerId,
       libp2p,
       { pingKeepAlive, relayKeepAlive },
-      pubsubTopics,
+      this.pubsubTopics,
       this.relay
     );
 

packages/interfaces/src/filter.ts

@@ -1,6 +1,6 @@
 import type { PeerId } from "@libp2p/interface/peer-id";
 
-import type { IDecodedMessage, IDecoder } from "./message.js";
+import type { IDecodedMessage, IDecoder, SingleShardInfo } from "./message.js";
 import type { ContentTopic } from "./misc.js";
 import type { Callback, IBaseProtocol } from "./protocols.js";
 import type { IReceiver } from "./receiver.js";
@@ -25,7 +25,7 @@ export interface IFilterSubscription {
 export type IFilter = IReceiver &
   IBaseProtocol & {
     createSubscription(
-      pubsubTopic?: string,
+      pubsubTopicShardInfo?: SingleShardInfo,
       peerId?: PeerId
     ): Promise<IFilterSubscription>;
   };

packages/interfaces/src/message.ts

@@ -1,5 +1,10 @@
 import type { PubsubTopic } from "./misc.js";
 
+export interface SingleShardInfo {
+  cluster: number;
+  index: number;
+}
+
 export interface IRateLimitProof {
   proof: Uint8Array;
   merkleRoot: Uint8Array;
@@ -38,7 +43,7 @@ export interface IMetaSetter {
 }
 
 export interface EncoderOptions {
-  pubsubTopic?: PubsubTopic;
+  pubsubTopicShardInfo?: SingleShardInfo;
   /** The content topic to set on outgoing messages. */
   contentTopic: string;
   /**

packages/interfaces/src/protocols.ts

@@ -2,9 +2,9 @@ import type { Libp2p } from "@libp2p/interface";
 import type { PeerId } from "@libp2p/interface/peer-id";
 import type { Peer, PeerStore } from "@libp2p/interface/peer-store";
 
+import type { ShardInfo } from "./enr.js";
 import type { CreateLibp2pOptions } from "./libp2p.js";
 import type { IDecodedMessage } from "./message.js";
-import type { PubsubTopic } from "./misc.js";
 
 export enum Protocols {
   Relay = "relay",
@@ -23,9 +23,9 @@ export interface IBaseProtocol {
 
 export type ProtocolCreateOptions = {
   /**
-   * Waku supports usage of multiple pubsub topics, but this is still in early stages.
-   * Waku implements sharding to achieve scalability
-   * The format of the sharded topic is `/waku/2/rs/<shard_cluster_index>/<shard_number>`
+   * Waku supports usage of multiple pubsub topics. This is achieved through static sharding for now, and auto-sharding in the future.
+   * The format to specify a shard is:
+   * clusterId: number, shards: number[]
    * To learn more about the sharding specifications implemented, see [Relay Sharding](https://rfc.vac.dev/spec/51/).
    * The Pubsub Topic to use. Defaults to {@link @waku/core!DefaultPubsubTopic }.
    *
@@ -39,7 +39,7 @@ export type ProtocolCreateOptions = {
    * See [Waku v2 Topic Usage Recommendations](https://rfc.vac.dev/spec/23/) for details.
    *
    */
-  pubsubTopics?: PubsubTopic[];
+  shardInfo?: ShardInfo;
   /**
    * You can pass options to the `Libp2p` instance used by {@link @waku/core!WakuNode} using the `libp2p` property.
    * This property is the same type as the one passed to [`Libp2p.create`](https://github.com/libp2p/js-libp2p/blob/master/doc/API.md#create)

packages/message-encryption/src/ecies.ts

@@ -7,10 +7,11 @@ import type {
   IMessage,
   IMetaSetter,
   IProtoMessage,
-  PubsubTopic
+  PubsubTopic,
+  SingleShardInfo
 } from "@waku/interfaces";
 import { WakuMessage } from "@waku/proto";
-import { Logger } from "@waku/utils";
+import { Logger, singleShardInfoToPubsubTopic } from "@waku/utils";
 
 import { generatePrivateKey } from "./crypto/utils.js";
 import { DecodedMessage } from "./decoded_message.js";
@@ -98,15 +99,17 @@ export interface EncoderOptions extends BaseEncoderOptions {
  * in [26/WAKU2-PAYLOAD](https://rfc.vac.dev/spec/26/).
  */
 export function createEncoder({
-  pubsubTopic = DefaultPubsubTopic,
+  pubsubTopicShardInfo,
   contentTopic,
   publicKey,
   sigPrivKey,
   ephemeral = false,
   metaSetter
 }: EncoderOptions): Encoder {
   return new Encoder(
-    pubsubTopic,
+    pubsubTopicShardInfo?.index
+      ? singleShardInfoToPubsubTopic(pubsubTopicShardInfo)
+      : DefaultPubsubTopic,
     contentTopic,
     publicKey,
     sigPrivKey,
@@ -194,7 +197,13 @@ class Decoder extends DecoderV0 implements IDecoder<DecodedMessage> {
 export function createDecoder(
   contentTopic: string,
   privateKey: Uint8Array,
-  pubsubTopic: PubsubTopic = DefaultPubsubTopic
+  pubsubTopicShardInfo?: SingleShardInfo
 ): Decoder {
-  return new Decoder(pubsubTopic, contentTopic, privateKey);
+  return new Decoder(
+    pubsubTopicShardInfo?.index
+      ? singleShardInfoToPubsubTopic(pubsubTopicShardInfo)
+      : DefaultPubsubTopic,
+    contentTopic,
+    privateKey
+  );
 }

packages/message-encryption/src/symmetric.ts

@@ -7,10 +7,11 @@ import type {
   IMessage,
   IMetaSetter,
   IProtoMessage,
-  PubsubTopic
+  PubsubTopic,
+  SingleShardInfo
 } from "@waku/interfaces";
 import { WakuMessage } from "@waku/proto";
-import { Logger } from "@waku/utils";
+import { Logger, singleShardInfoToPubsubTopic } from "@waku/utils";
 
 import { generateSymmetricKey } from "./crypto/utils.js";
 import { DecodedMessage } from "./decoded_message.js";
@@ -98,15 +99,17 @@ export interface EncoderOptions extends BaseEncoderOptions {
  * in [26/WAKU2-PAYLOAD](https://rfc.vac.dev/spec/26/).
  */
 export function createEncoder({
-  pubsubTopic = DefaultPubsubTopic,
+  pubsubTopicShardInfo,
   contentTopic,
   symKey,
   sigPrivKey,
   ephemeral = false,
   metaSetter
 }: EncoderOptions): Encoder {
   return new Encoder(
-    pubsubTopic,
+    pubsubTopicShardInfo?.index
+      ? singleShardInfoToPubsubTopic(pubsubTopicShardInfo)
+      : DefaultPubsubTopic,
     contentTopic,
     symKey,
     sigPrivKey,
@@ -194,7 +197,13 @@ class Decoder extends DecoderV0 implements IDecoder<DecodedMessage> {
 export function createDecoder(
   contentTopic: string,
   symKey: Uint8Array,
-  pubsubTopic: PubsubTopic = DefaultPubsubTopic
+  pubsubTopicShardInfo?: SingleShardInfo
 ): Decoder {
-  return new Decoder(pubsubTopic, contentTopic, symKey);
+  return new Decoder(
+    pubsubTopicShardInfo?.index
+      ? singleShardInfoToPubsubTopic(pubsubTopicShardInfo)
+      : DefaultPubsubTopic,
+    contentTopic,
+    symKey
+  );
 }