feat: use the lowest latency peer for light protocols

packages/core/src/index.ts

@@ -15,15 +15,11 @@ export * as waku_filter from "./lib/filter/index.js";
 export { wakuFilter, FilterCodecs } from "./lib/filter/index.js";
 
 export * as waku_light_push from "./lib/light_push/index.js";
-export { wakuLightPush, LightPushCodec } from "./lib/light_push/index.js";
+export { wakuLightPush } from "./lib/light_push/index.js";
 
 export * as waku_store from "./lib/store/index.js";
-export {
-  PageDirection,
-  wakuStore,
-  StoreCodec,
-  createCursor
-} from "./lib/store/index.js";
+
+export { PageDirection, wakuStore, createCursor } from "./lib/store/index.js";
 
 export { waitForRemotePeer } from "./lib/wait_for_remote_peer.js";
 

packages/core/src/lib/base_protocol.spec.ts

@@ -0,0 +1,143 @@
+import { Peer } from "@libp2p/interface/peer-store";
+import type { Tag } from "@libp2p/interface/peer-store";
+import { createSecp256k1PeerId } from "@libp2p/peer-id-factory";
+import { Tags } from "@waku/interfaces";
+import { filterPeers } from "@waku/utils";
+import { expect } from "chai";
+
+describe("getPeers function", function () {
+  it("should return all peers when numPeers is 0", async function () {
+    const peer1 = await createSecp256k1PeerId();
+    const peer2 = await createSecp256k1PeerId();
+    const peer3 = await createSecp256k1PeerId();
+
+    const mockPeers = [
+      {
+        id: peer1,
+        tags: new Map<string, Tag>([[Tags.BOOTSTRAP, { value: 100 }]])
+      },
+      {
+        id: peer2,
+        tags: new Map<string, Tag>([[Tags.BOOTSTRAP, { value: 100 }]])
+      },
+      {
+        id: peer3,
+        tags: new Map<string, Tag>([[Tags.BOOTSTRAP, { value: 100 }]])
+      }
+    ] as unknown as Peer[];
+
+    const result = await filterPeers(mockPeers, 0, 10);
+    expect(result.length).to.deep.equal(mockPeers.length);
+  });
+
+  it("should return all non-bootstrap peers and no bootstrap peer when numPeers is 0 and maxBootstrapPeers is 0", async function () {
+    const peer1 = await createSecp256k1PeerId();
+    const peer2 = await createSecp256k1PeerId();
+    const peer3 = await createSecp256k1PeerId();
+    const peer4 = await createSecp256k1PeerId();
+
+    const mockPeers = [
+      {
+        id: peer1,
+        tags: new Map<string, Tag>([[Tags.BOOTSTRAP, { value: 100 }]])
+      },
+      {
+        id: peer2,
+        tags: new Map<string, Tag>([[Tags.BOOTSTRAP, { value: 100 }]])
+      },
+      {
+        id: peer3,
+        tags: new Map<string, Tag>([[Tags.PEER_EXCHANGE, { value: 100 }]])
+      },
+      {
+        id: peer4,
+        tags: new Map<string, Tag>([[Tags.PEER_EXCHANGE, { value: 100 }]])
+      }
+    ] as unknown as Peer[];
+
+    const result = await filterPeers(mockPeers, 0, 0);
+
+    // result should have no bootstrap peers, and a total of 2 peers
+    expect(result.length).to.equal(2);
+    expect(
+      result.filter((peer: Peer) => peer.tags.has(Tags.BOOTSTRAP)).length
+    ).to.equal(0);
+  });
+
+  it("should return one bootstrap peer, and all non-boostrap peers, when numPeers is 0 & maxBootstrap is 1", async function () {
+    const peer1 = await createSecp256k1PeerId();
+    const peer2 = await createSecp256k1PeerId();
+    const peer3 = await createSecp256k1PeerId();
+    const peer4 = await createSecp256k1PeerId();
+    const peer5 = await createSecp256k1PeerId();
+
+    const mockPeers = [
+      {
+        id: peer1,
+        tags: new Map<string, Tag>([[Tags.BOOTSTRAP, { value: 100 }]])
+      },
+      {
+        id: peer2,
+        tags: new Map<string, Tag>([[Tags.BOOTSTRAP, { value: 100 }]])
+      },
+      {
+        id: peer3,
+        tags: new Map<string, Tag>([[Tags.PEER_EXCHANGE, { value: 100 }]])
+      },
+      {
+        id: peer4,
+        tags: new Map<string, Tag>([[Tags.PEER_EXCHANGE, { value: 100 }]])
+      },
+      {
+        id: peer5,
+        tags: new Map<string, Tag>([[Tags.PEER_EXCHANGE, { value: 100 }]])
+      }
+    ] as unknown as Peer[];
+
+    const result = await filterPeers(mockPeers, 0, 1);
+
+    // result should have 1 bootstrap peers, and a total of 4 peers
+    expect(result.length).to.equal(4);
+    expect(
+      result.filter((peer: Peer) => peer.tags.has(Tags.BOOTSTRAP)).length
+    ).to.equal(1);
+  });
+
+  it("should return only bootstrap peers up to maxBootstrapPeers", async function () {
+    const peer1 = await createSecp256k1PeerId();
+    const peer2 = await createSecp256k1PeerId();
+    const peer3 = await createSecp256k1PeerId();
+    const peer4 = await createSecp256k1PeerId();
+    const peer5 = await createSecp256k1PeerId();
+
+    const mockPeers = [
+      {
+        id: peer1,
+        tags: new Map<string, Tag>([[Tags.BOOTSTRAP, { value: 100 }]])
+      },
+      {
+        id: peer2,
+        tags: new Map<string, Tag>([[Tags.BOOTSTRAP, { value: 100 }]])
+      },
+      {
+        id: peer3,
+        tags: new Map<string, Tag>([[Tags.BOOTSTRAP, { value: 100 }]])
+      },
+      {
+        id: peer4,
+        tags: new Map<string, Tag>([[Tags.PEER_EXCHANGE, { value: 100 }]])
+      },
+      {
+        id: peer5,
+        tags: new Map<string, Tag>([[Tags.PEER_EXCHANGE, { value: 100 }]])
+      }
+    ] as unknown as Peer[];
+
+    const result = await filterPeers(mockPeers, 5, 2);
+
+    // check that result has at least 2 bootstrap peers and no more than 5 peers
+    expect(result.length).to.be.at.least(2);
+    expect(result.length).to.be.at.most(5);
+    expect(result.filter((peer: Peer) => peer.tags.has(Tags.BOOTSTRAP)).length);
+  });
+});

packages/core/src/lib/base_protocol.ts

@@ -3,8 +3,10 @@ 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 { filterPeers } from "@waku/utils";
 import { getPeersForProtocol, selectPeerForProtocol } from "@waku/utils/libp2p";
 
+import { KeepAliveManager } from "./keep_alive_manager.js";
 import { StreamManager } from "./stream_manager.js";
 
 /**
@@ -53,11 +55,41 @@ export class BaseProtocol implements IBaseProtocol {
   }
 
   protected async getPeer(peerId?: PeerId): Promise<Peer> {
+    const { peerPings } = KeepAliveManager.getInstance();
     const { peer } = await selectPeerForProtocol(
       this.peerStore,
+      peerPings,
       [this.multicodec],
       peerId
     );
     return peer;
   }
+
+  /**
+   * Retrieves a list of peers based on the specified criteria.
+   *
+   * @param numPeers - The total number of peers to retrieve. If 0, all peers are returned.
+   * @param maxBootstrapPeers - The maximum number of bootstrap peers to retrieve.
+   * @returns A Promise that resolves to an array of peers based on the specified criteria.
+   */
+  protected async getPeers(
+    {
+      numPeers,
+      maxBootstrapPeers
+    }: {
+      numPeers: number;
+      maxBootstrapPeers: number;
+    } = {
+      maxBootstrapPeers: 1,
+      numPeers: 0
+    }
+  ): Promise<Peer[]> {
+    // Retrieve all peers that support the protocol
+    const allPeersForProtocol = await getPeersForProtocol(this.peerStore, [
+      this.multicodec
+    ]);
+
+    // Filter the peers based on the specified criteria
+    return filterPeers(allPeersForProtocol, numPeers, maxBootstrapPeers);
+  }
 }

packages/core/src/lib/connection_manager.ts

@@ -116,7 +116,11 @@ export class ConnectionManager
       ...options
     };
 
-    this.keepAliveManager = new KeepAliveManager(keepAliveOptions, relay);
+    this.keepAliveManager = KeepAliveManager.createInstance(
+      this.libp2p.services.ping,
+      keepAliveOptions,
+      relay
+    );
 
     this.run()
       .then(() => log(`Connection Manager is now running`))
@@ -340,7 +344,7 @@ export class ConnectionManager
       void (async () => {
         const peerId = evt.detail;
 
-        this.keepAliveManager.start(peerId, this.libp2p.services.ping);
+        this.keepAliveManager.start(peerId);
 
         const isBootstrap = (await this.getTagNamesForPeer(peerId)).includes(
           Tags.BOOTSTRAP

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

@@ -1,5 +1,4 @@
 import { Stream } from "@libp2p/interface/connection";
-import type { PeerId } from "@libp2p/interface/peer-id";
 import type { Peer } from "@libp2p/interface/peer-store";
 import type { IncomingStreamData } from "@libp2p/interface-internal/registrar";
 import type {
@@ -14,7 +13,6 @@ import type {
   Libp2p,
   PeerIdStr,
   ProtocolCreateOptions,
-  ProtocolOptions,
   PubSubTopic,
   Unsubscribe
 } from "@waku/interfaces";
@@ -228,6 +226,7 @@ class Subscription {
 class Filter extends BaseProtocol implements IReceiver {
   private readonly options: ProtocolCreateOptions;
   private activeSubscriptions = new Map<string, Subscription>();
+  private readonly NUM_PEERS_PROTOCOL = 1;
 
   private getActiveSubscription(
     pubSubTopic: PubSubTopic,
@@ -257,14 +256,16 @@ class Filter extends BaseProtocol implements IReceiver {
     this.options = options ?? {};
   }
 
-  async createSubscription(
-    pubSubTopic?: string,
-    peerId?: PeerId
-  ): Promise<Subscription> {
+  async createSubscription(pubSubTopic?: string): Promise<Subscription> {
     const _pubSubTopic =
       pubSubTopic ?? this.options.pubSubTopic ?? DefaultPubSubTopic;
 
-    const peer = await this.getPeer(peerId);
+    const peer = (
+      await this.getPeers({
+        maxBootstrapPeers: 1,
+        numPeers: this.NUM_PEERS_PROTOCOL
+      })
+    )[0];
 
     const subscription =
       this.getActiveSubscription(_pubSubTopic, peer.id.toString()) ??
@@ -278,10 +279,9 @@ class Filter extends BaseProtocol implements IReceiver {
   }
 
   public toSubscriptionIterator<T extends IDecodedMessage>(
-    decoders: IDecoder<T> | IDecoder<T>[],
-    opts?: ProtocolOptions | undefined
+    decoders: IDecoder<T> | IDecoder<T>[]
   ): Promise<IAsyncIterator<T>> {
-    return toAsyncIterator(this, decoders, opts);
+    return toAsyncIterator(this, decoders);
   }
 
   /**
@@ -301,10 +301,9 @@ class Filter extends BaseProtocol implements IReceiver {
    */
   async subscribe<T extends IDecodedMessage>(
     decoders: IDecoder<T> | IDecoder<T>[],
-    callback: Callback<T>,
-    opts?: ProtocolOptions
+    callback: Callback<T>
   ): Promise<Unsubscribe> {
-    const subscription = await this.createSubscription(undefined, opts?.peerId);
+    const subscription = await this.createSubscription();
 
     await subscription.subscribe(decoders, callback);