feat(static-sharding)!: allow multiple pubSubTopics (#1586)

* `ProtocolCreateOptions` now has `pubSubTopic` as `pubSubTopic[]`

* chore: update encoder & decoder to support `PubSubTopic`

* feat(protocols): allow multiple `PubSubTopic[]`

* feat(relay): allow multiple `PubSubTopic[]`

* chore(tests): update for new API

* chore: minor fixes

* chore: make store more robust

* fix(relay): correctly set types

* chore(address comments): update terminology around configured pubsub topics

* chore(address comments): minor refactoring

* chore(relay): split `subscribe` into smaller functions for readability & modularity

* chore(address comments): refactor `waitForGossipSubPeerInMesh`

* chore(store): only allow to query one `pubSubTopic`

* fix: `store` bug

* feat(tests): add some basic tests

* sharding utils

* address comments

* feat(relay): re-add API for `getMeshPeers`

* update error message

Co-authored-by: fryorcraken <110212804+fryorcraken@users.noreply.github.com>

* refactor for new API

* feat: simplify handling of observers (#1614)

* refactor: simplify handling of observers

* refactor: Remove redundant PubSubTopic from Observer

* use `??` instead of `||`

* update `pubsubTopic` to `pubSubTopic`

* update `interval` typo

* change occurence of `pubsubTopic` to `pubSubTopic`

* relay: rm `getAllMeshPeers` and make `pubSubTopics` public

* relay: use `push_or_init_map` and move to `utils`

* fix: update API for tests

* fix: relay waitForRemotePeer

---------

Co-authored-by: fryorcraken <110212804+fryorcraken@users.noreply.github.com>

Author: danisharora099

.cspell.json

@@ -39,6 +39,7 @@
     "exponentiate",
     "extip",
     "fanout",
+    "sharded",
     "floodsub",
     "fontsource",
     "globby",

package-lock.json

@@ -17,7 +17,11 @@ import type {
   Unsubscribe
 } from "@waku/interfaces";
 import { WakuMessage } from "@waku/proto";
-import { groupByContentTopic, toAsyncIterator } from "@waku/utils";
+import {
+  ensurePubsubTopicIsConfigured,
+  groupByContentTopic,
+  toAsyncIterator
+} from "@waku/utils";
 import debug from "debug";
 import all from "it-all";
 import * as lp from "it-length-prefixed";
@@ -230,7 +234,7 @@ class Subscription {
 }
 
 class Filter extends BaseProtocol implements IReceiver {
-  private readonly options: ProtocolCreateOptions;
+  private readonly pubSubTopics: PubSubTopic[] = [];
   private activeSubscriptions = new Map<string, Subscription>();
   private readonly NUM_PEERS_PROTOCOL = 1;
 
@@ -253,19 +257,22 @@ class Filter extends BaseProtocol implements IReceiver {
   constructor(libp2p: Libp2p, options?: ProtocolCreateOptions) {
     super(FilterCodecs.SUBSCRIBE, libp2p.components);
 
+    this.pubSubTopics = options?.pubSubTopics || [DefaultPubSubTopic];
+
     libp2p.handle(FilterCodecs.PUSH, this.onRequest.bind(this)).catch((e) => {
       log("Failed to register ", FilterCodecs.PUSH, e);
     });
 
     this.activeSubscriptions = new Map();
-
-    this.options = options ?? {};
   }
 
-  async createSubscription(pubSubTopic?: string): Promise<Subscription> {
-    const _pubSubTopic =
-      pubSubTopic ?? this.options.pubSubTopic ?? DefaultPubSubTopic;
+  async createSubscription(
+    pubSubTopic: string = DefaultPubSubTopic
+  ): Promise<Subscription> {
+    ensurePubsubTopicIsConfigured(pubSubTopic, this.pubSubTopics);
 
+    //TODO: get a relevant peer for the topic/shard
+    // https://github.com/waku-org/js-waku/pull/1586#discussion_r1336428230
     const peer = (
       await this.getPeers({
         maxBootstrapPeers: 1,
@@ -274,11 +281,11 @@ class Filter extends BaseProtocol implements IReceiver {
     )[0];
 
     const subscription =
-      this.getActiveSubscription(_pubSubTopic, peer.id.toString()) ??
+      this.getActiveSubscription(pubSubTopic, peer.id.toString()) ??
       this.setActiveSubscription(
-        _pubSubTopic,
+        pubSubTopic,
         peer.id.toString(),
-        new Subscription(_pubSubTopic, peer, this.getStream.bind(this, peer))
+        new Subscription(pubSubTopic, peer, this.getStream.bind(this, peer))
       );
 
     return subscription;

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

@@ -1,6 +1,6 @@
 import type { PeerId } from "@libp2p/interface/peer-id";
 import type { PeerStore } from "@libp2p/interface/peer-store";
-import type { IRelay } from "@waku/interfaces";
+import type { IRelay, PeerIdStr } from "@waku/interfaces";
 import type { KeepAliveOptions } from "@waku/interfaces";
 import { utf8ToBytes } from "@waku/utils/bytes";
 import debug from "debug";
@@ -13,7 +13,7 @@ const log = debug("waku:keep-alive");
 
 export class KeepAliveManager {
   private pingKeepAliveTimers: Map<string, ReturnType<typeof setInterval>>;
-  private relayKeepAliveTimers: Map<PeerId, ReturnType<typeof setInterval>>;
+  private relayKeepAliveTimers: Map<PeerId, ReturnType<typeof setInterval>[]>;
   private options: KeepAliveOptions;
   private relay?: IRelay;
 
@@ -66,17 +66,12 @@ export class KeepAliveManager {
 
     const relay = this.relay;
     if (relay && relayPeriodSecs !== 0) {
-      const encoder = createEncoder({
-        contentTopic: RelayPingContentTopic,
-        ephemeral: true
-      });
-      const interval = setInterval(() => {
-        log("Sending Waku Relay ping message");
-        relay
-          .send(encoder, { payload: new Uint8Array([1]) })
-          .catch((e) => log("Failed to send relay ping", e));
-      }, relayPeriodSecs * 1000);
-      this.relayKeepAliveTimers.set(peerId, interval);
+      const intervals = this.scheduleRelayPings(
+        relay,
+        relayPeriodSecs,
+        peerId.toString()
+      );
+      this.relayKeepAliveTimers.set(peerId, intervals);
     }
   }
 
@@ -89,7 +84,7 @@ export class KeepAliveManager {
     }
 
     if (this.relayKeepAliveTimers.has(peerId)) {
-      clearInterval(this.relayKeepAliveTimers.get(peerId));
+      this.relayKeepAliveTimers.get(peerId)?.map(clearInterval);
       this.relayKeepAliveTimers.delete(peerId);
     }
   }
@@ -105,4 +100,32 @@ export class KeepAliveManager {
     this.pingKeepAliveTimers.clear();
     this.relayKeepAliveTimers.clear();
   }
+
+  private scheduleRelayPings(
+    relay: IRelay,
+    relayPeriodSecs: number,
+    peerIdStr: PeerIdStr
+  ): NodeJS.Timeout[] {
+    // send a ping message to each PubSubTopic the peer is part of
+    const intervals: NodeJS.Timeout[] = [];
+    for (const topic of relay.pubSubTopics) {
+      const meshPeers = relay.getMeshPeers(topic);
+      if (!meshPeers.includes(peerIdStr)) continue;
+
+      const encoder = createEncoder({
+        pubSubTopic: topic,
+        contentTopic: RelayPingContentTopic,
+        ephemeral: true
+      });
+      const interval = setInterval(() => {
+        log("Sending Waku Relay ping message");
+        relay
+          .send(encoder, { payload: new Uint8Array([1]) })
+          .catch((e) => log("Failed to send relay ping", e));
+      }, relayPeriodSecs * 1000);
+      intervals.push(interval);
+    }
+
+    return intervals;
+  }
 }

packages/core/src/lib/keep_alive_manager.ts

@@ -6,11 +6,12 @@ import {
   IMessage,
   Libp2p,
   ProtocolCreateOptions,
+  PubSubTopic,
   SendError,
   SendResult
 } from "@waku/interfaces";
 import { PushResponse } from "@waku/proto";
-import { isSizeValid } from "@waku/utils";
+import { ensurePubsubTopicIsConfigured, isSizeValid } from "@waku/utils";
 import debug from "debug";
 import all from "it-all";
 import * as lp from "it-length-prefixed";
@@ -41,12 +42,12 @@ type PreparePushMessageResult =
  * Implements the [Waku v2 Light Push protocol](https://rfc.vac.dev/spec/19/).
  */
 class LightPush extends BaseProtocol implements ILightPush {
-  options: ProtocolCreateOptions;
+  private readonly pubSubTopics: PubSubTopic[];
   private readonly NUM_PEERS_PROTOCOL = 1;
 
   constructor(libp2p: Libp2p, options?: ProtocolCreateOptions) {
     super(LightPushCodec, libp2p.components);
-    this.options = options || {};
+    this.pubSubTopics = options?.pubSubTopics ?? [DefaultPubSubTopic];
   }
 
   private async preparePushMessage(
@@ -82,7 +83,9 @@ class LightPush extends BaseProtocol implements ILightPush {
   }
 
   async send(encoder: IEncoder, message: IMessage): Promise<SendResult> {
-    const { pubSubTopic = DefaultPubSubTopic } = this.options;
+    const { pubSubTopic } = encoder;
+    ensurePubsubTopicIsConfigured(pubSubTopic, this.pubSubTopics);
+
     const recipients: PeerId[] = [];
 
     const { query, error: preparationError } = await this.preparePushMessage(
@@ -98,6 +101,7 @@ class LightPush extends BaseProtocol implements ILightPush {
       };
     }
 
+    //TODO: get a relevant peer for the topic/shard
     const peers = await this.getPeers({
       maxBootstrapPeers: 1,
       numPeers: this.NUM_PEERS_PROTOCOL

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

@@ -6,11 +6,14 @@ import type {
   IMessage,
   IMetaSetter,
   IProtoMessage,
-  IRateLimitProof
+  IRateLimitProof,
+  PubSubTopic
 } from "@waku/interfaces";
 import { proto_message as proto } from "@waku/proto";
 import debug from "debug";
 
+import { DefaultPubSubTopic } from "../constants.js";
+
 const log = debug("waku:message:version-0");
 const OneMillion = BigInt(1_000_000);
 
@@ -73,6 +76,7 @@ export class Encoder implements IEncoder {
   constructor(
     public contentTopic: string,
     public ephemeral: boolean = false,
+    public pubSubTopic: PubSubTopic,
     public metaSetter?: IMetaSetter
   ) {
     if (!contentTopic || contentTopic === "") {
@@ -115,15 +119,19 @@ export class Encoder implements IEncoder {
  * messages.
  */
 export function createEncoder({
+  pubSubTopic = DefaultPubSubTopic,
   contentTopic,
   ephemeral,
   metaSetter
 }: EncoderOptions): Encoder {
-  return new Encoder(contentTopic, ephemeral, metaSetter);
+  return new Encoder(contentTopic, ephemeral, pubSubTopic, metaSetter);
 }
 
 export class Decoder implements IDecoder<DecodedMessage> {
-  constructor(public contentTopic: string) {
+  constructor(
+    public pubSubTopic: PubSubTopic,
+    public contentTopic: string
+  ) {
     if (!contentTopic || contentTopic === "") {
       throw new Error("Content topic must be specified");
     }
@@ -173,6 +181,9 @@ export class Decoder implements IDecoder<DecodedMessage> {
  *
  * @param contentTopic The resulting decoder will only decode messages with this content topic.
  */
-export function createDecoder(contentTopic: string): Decoder {
-  return new Decoder(contentTopic);
+export function createDecoder(
+  contentTopic: string,
+  pubsubTopic: PubSubTopic = DefaultPubSubTopic
+): Decoder {
+  return new Decoder(pubsubTopic, contentTopic);
 }