feat: use the lowest latency peer for light protocols

packages/core/src/lib/base_protocol.ts

@@ -46,6 +46,10 @@ export class BaseProtocol implements IBaseProtocol {
     const { peer } = await selectPeerForProtocol(
       this.peerStore,
       [this.multicodec],
+      {
+        registrar: this.components.registrar,
+        connectionManager: this.components.connectionManager
+      },
       peerId
     );
     return peer;

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

@@ -257,6 +257,19 @@ class Filter extends BaseProtocol implements IReceiver {
     this.options = options ?? {};
   }
 
+  /**
+   * Creates a subscription for a given pub/sub topic and peer.
+   *
+   * @param pubSubTopic - The pub/sub topic for the subscription. If not provided,
+   *                      it defaults to the protocol's configured topic or the default topic.
+   * @param peerId - The ID of the peer to use for Filter subscribe. If not specified,
+   *                 the peer with the lowest latency from the connected peers is used.
+   *
+   * @returns A promise that resolves to a `Subscription` object for the given topic and peer.
+   *
+   * @throws {Error} - Throws an error if there's an issue creating the subscription.
+   *
+   */
   async createSubscription(
     pubSubTopic?: string,
     peerId?: PeerId

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

@@ -64,6 +64,20 @@ class LightPush extends BaseProtocol implements ILightPush {
     return { query };
   }
 
+  /**
+   * Sends a message using the Waku v2 Light Push protocol.
+   *
+   * @param encoder - The encoder to use for encoding the message.
+   * @param message - The message to be sent.
+   * @param opts - Optional protocol options.
+   * @param opts.peerId - The ID of the peer to use for LightPush send. If not specified,
+   *                      the peer with the lowest latency from the connected peers is used.
+   *
+   * @returns A promise that resolves to a `SendResult` object containing the recipients
+   *          of the message and any potential error that occurred during the send operation.
+   *
+   * @throws {SendError} - Throws an error if there's an issue with sending the message.
+   */
   async send(
     encoder: IEncoder,
     message: IMessage,

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

@@ -41,7 +41,7 @@ export interface TimeFilter {
 
 export interface QueryOptions {
   /**
-   * The peer to query. If undefined, a pseudo-random peer is selected from the connected Waku Store peers.
+   * The peer to query. If undefined, the peer with the lowest latency is selected from the connected Waku Store peers.
    */
   peerId?: PeerId;
   /**
@@ -198,18 +198,20 @@ class Store extends BaseProtocol implements IStore {
   }
 
   /**
-   * Do a query to a Waku Store to retrieve historical/missed messages.
+   * Queries a Waku Store to retrieve historical or missed messages using a generator.
    *
-   * This is a generator, useful if you want most control on how messages
-   * are processed.
+   * This generator provides granular control over how messages are processed.
+   * Messages returned by the remote Waku node are ordered as follows:
+   * - Within a page, messages are ordered from the oldest to the most recent.
+   * - The direction of pages depends on the {@link QueryOptions.pageDirection}.
+   * @params options.peerId - The ID of the peer to use for the Store query. If not specified,
+   *                         the peer with the lowest latency from the connected peers is used.
    *
-   * The order of the messages returned by the remote Waku node SHOULD BE
-   * as follows:
-   * - within a page, messages SHOULD be ordered from oldest to most recent
-   * - pages direction depends on { @link QueryOptions.pageDirection }
-   * @throws If not able to reach a Waku Store peer to query,
-   * or if an error is encountered when processing the reply,
-   * or if two decoders with the same content topic are passed.
+   * @yields An array of promises that resolve to decoded messages.
+   *
+   * @throws {Error} - Throws an error if unable to reach a Waku Store peer,
+   *                   if an error occurs while processing the reply,
+   *                   or if two decoders with the same content topic are passed.
    */
   async *queryGenerator<T extends IDecodedMessage>(
     decoders: IDecoder<T>[],

packages/interfaces/src/misc.ts

@@ -1,3 +1,4 @@
+import { Libp2pComponents } from "./libp2p.js";
 import type { IDecodedMessage } from "./message.js";
 
 export interface IAsyncIterator<T extends IDecodedMessage> {
@@ -11,3 +12,8 @@ export type PubSubTopic = string;
 export type ContentTopic = string;
 
 export type PeerIdStr = string;
+
+export interface PingServiceComponents {
+  registrar: Libp2pComponents["registrar"];
+  connectionManager: Libp2pComponents["connectionManager"];
+}

packages/sdk/src/create.ts

@@ -175,7 +175,7 @@ export async function defaultLibp2p(
       identify: identifyService({
         agentVersion: userAgent ?? DefaultUserAgent
       }),
-      ping: pingService(),
+      ping: pingService({ maxOutboundStreams: 2 }),
       ...pubsubService,
       ...options?.services
     }

packages/utils/src/libp2p/index.ts

@@ -1,11 +1,14 @@
 import type { Connection } from "@libp2p/interface/connection";
 import type { PeerId } from "@libp2p/interface/peer-id";
 import type { Peer, PeerStore } from "@libp2p/interface/peer-store";
+import type { PingServiceComponents } from "@waku/interfaces";
 import debug from "debug";
+import type { PingService } from "libp2p/ping";
+import { pingService as libp2pPingService } from "libp2p/ping";
 
 const log = debug("waku:libp2p-utils");
-
 /**
+ * @deprecated uses fastest peer selection instead
  * Returns a pseudo-random peer that supports the given protocol.
  * Useful for protocols such as store and light push
  */
@@ -16,6 +19,30 @@ export function selectRandomPeer(peers: Peer[]): Peer | undefined {
   return peers[index];
 }
 
+/**
+ *
+ * @param ping The libp2p ping service's ping function
+ * @param peerStore The libp2p peer store
+ * @param protocols The protocols that the peer must support
+ * @returns The peer with the lowest latency that supports the given protocols
+ */
+
+async function selectFastestPeer(
+  ping: PingService["ping"],
+  peers: Peer[]
+): Promise<Peer> {
+  const peerLatencies = await Promise.all(
+    peers.map(async (peer) => {
+      const latency = await ping(peer.id);
+      return { peer, latency };
+    })
+  );
+
+  peerLatencies.sort((a, b) => a.latency - b.latency);
+
+  return peerLatencies[0].peer;
+}
+
 /**
  * Returns the list of peers that supports the given protocol.
  */
@@ -35,12 +62,23 @@ export async function getPeersForProtocol(
   return peers;
 }
 
+/**
+ * Returns a peer that supports the given protocol.
+ * If peerId is provided, it will be used to retrieve the peer's connection details.
+ * Otherwise, the peer with the lowest latency will be selected.
+ * @param peerStore The libp2p peer store
+ * @param protocols The protocols that the peer must support
+ * @param ping The libp2p ping service's ping function
+ * @param peerId The peerId of the peer to select
+ * @returns The peer and protocol that was selected
+ */
 export async function selectPeerForProtocol(
   peerStore: PeerStore,
   protocols: string[],
+  { registrar, connectionManager }: PingServiceComponents,
   peerId?: PeerId
 ): Promise<{ peer: Peer; protocol: string }> {
-  let peer;
+  let peer: Peer;
   if (peerId) {
     peer = await peerStore.get(peerId);
     if (!peer) {
@@ -50,7 +88,9 @@ export async function selectPeerForProtocol(
     }
   } else {
     const peers = await getPeersForProtocol(peerStore, protocols);
-    peer = selectRandomPeer(peers);
+    const pingService = libp2pPingService()({ connectionManager, registrar });
+    const { ping } = pingService;
+    peer = await selectFastestPeer(ping.bind(pingService), peers);
     if (!peer) {
       throw new Error(
         `Failed to find known peer that registers protocols: ${protocols}`