Merge pull request #1244 from waku-org/chore/peer-exchange

Author: fryorcraken

package-lock.json

@@ -10,6 +10,7 @@ import {
   ProtocolCreateOptions,
 } from "@waku/interfaces";
 import { proto_store as proto } from "@waku/proto";
+import { isDefined } from "@waku/utils";
 import { concat, utf8ToBytes } from "@waku/utils/bytes";
 import debug from "debug";
 import all from "it-all";
@@ -345,10 +346,6 @@ async function* paginate<T extends IDecodedMessage>(
   }
 }
 
-export function isDefined<T>(msg: T | undefined): msg is T {
-  return !!msg;
-}
-
 export async function createCursor(
   message: IDecodedMessage,
   pubsubTopic: string = DefaultPubSubTopic

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

@@ -6,7 +6,7 @@ import { IEnr } from "./enr.js";
 import { PointToPointProtocol } from "./protocols.js";
 
 export interface IPeerExchange extends PointToPointProtocol {
-  query(params: PeerExchangeQueryParams): Promise<PeerInfo[]>;
+  query(params: PeerExchangeQueryParams): Promise<PeerInfo[] | undefined>;
 }
 
 export interface PeerExchangeQueryParams {

packages/interfaces/src/peer_exchange.ts

@@ -54,6 +54,7 @@
     "@libp2p/interfaces": "^3.3.1",
     "@waku/enr": "0.0.6",
     "@waku/proto": "0.0.3",
+    "@waku/utils": "*",
     "debug": "^4.3.4",
     "it-all": "^2.0.0",
     "it-length-prefixed": "^8.0.4",

packages/peer-exchange/package.json

@@ -7,6 +7,7 @@ import type {
   PeerExchangeQueryParams,
   PeerInfo,
 } from "@waku/interfaces";
+import { isDefined } from "@waku/utils";
 import debug from "debug";
 import all from "it-all";
 import * as lp from "it-length-prefixed";
@@ -47,7 +48,9 @@ export class WakuPeerExchange extends BaseProtocol implements IPeerExchange {
   /**
    * Make a peer exchange query to a peer
    */
-  async query(params: PeerExchangeQueryParams): Promise<PeerInfo[]> {
+  async query(
+    params: PeerExchangeQueryParams
+  ): Promise<PeerInfo[] | undefined> {
     const { numPeers } = params;
 
     const rpcQuery = PeerExchangeRPC.createRequest({
@@ -72,28 +75,24 @@ export class WakuPeerExchange extends BaseProtocol implements IPeerExchange {
         bytes.append(chunk);
       });
 
-      const decoded = PeerExchangeRPC.decode(bytes).response;
+      const { response } = PeerExchangeRPC.decode(bytes);
 
-      if (!decoded) {
-        throw new Error("Failed to decode response");
+      if (!response) {
+        log("PeerExchangeRPC message did not contains a `response` field");
+        return;
       }
 
-      const enrs = await Promise.all(
-        decoded.peerInfos.map(
-          (peerInfo) => peerInfo.enr && EnrDecoder.fromRLP(peerInfo.enr)
-        )
+      return Promise.all(
+        response.peerInfos
+          .map((peerInfo) => peerInfo.enr)
+          .filter(isDefined)
+          .map(async (enr) => {
+            return { ENR: await EnrDecoder.fromRLP(enr) };
+          })
       );
-
-      const peerInfos = enrs.map((enr) => {
-        return {
-          ENR: enr,
-        };
-      });
-
-      return peerInfos;
     } catch (err) {
       log("Failed to decode push reply", err);
-      throw new Error("Failed to decode push reply");
+      return;
     }
   }
 }

packages/peer-exchange/src/waku_peer_exchange.ts

@@ -155,10 +155,15 @@ export class PeerExchangeDiscovery
       peerId,
     });
 
+    if (!peerInfos) {
+      log("Peer exchange query failed, no peer info returned");
+      return;
+    }
+
     for (const _peerInfo of peerInfos) {
       const { ENR } = _peerInfo;
       if (!ENR) {
-        log("no ENR");
+        log("No ENR in peerInfo object, skipping");
         continue;
       }
 

packages/peer-exchange/src/waku_peer_exchange_discovery.ts

@@ -4,6 +4,7 @@ import type { PeerId } from "@libp2p/interface-peer-id";
 import { peerIdFromString } from "@libp2p/peer-id";
 import { Multiaddr, multiaddr } from "@multiformats/multiaddr";
 import { DefaultPubSubTopic } from "@waku/core";
+import { isDefined } from "@waku/utils";
 import { bytesToHex, hexToBytes } from "@waku/utils/bytes";
 import appRoot from "app-root-path";
 import debug from "debug";
@@ -272,12 +273,6 @@ export class Nwaku {
   ): Promise<MessageRpcResponse[]> {
     this.checkProcess();
 
-    const isDefined = (
-      msg: MessageRpcResponse | undefined
-    ): msg is MessageRpcResponse => {
-      return !!msg;
-    };
-
     const msgs = await this.rpcCall<MessageRpcResponse[]>(
       "get_waku_v2_relay_v1_messages",
       [pubsubTopic]

packages/tests/src/nwaku.ts

@@ -4,6 +4,7 @@ import {
   getPredefinedBootstrapNodes,
 } from "@waku/core/lib/predefined_bootstrap_nodes";
 import { createLightNode } from "@waku/create";
+import { PeerInfo } from "@waku/interfaces";
 import type { LightNode } from "@waku/interfaces";
 import {
   PeerExchangeCodec,
@@ -117,9 +118,9 @@ describe("Peer Exchange", () => {
 
       const numPeersToRequest = 1;
 
-      const peerInfos = await peerExchange.query({
+      const peerInfos = (await peerExchange.query({
         numPeers: numPeersToRequest,
-      });
+      })) as PeerInfo[];
 
       expect(peerInfos.length).to.be.greaterThan(0);
       expect(peerInfos.length).to.be.lessThanOrEqual(numPeersToRequest);

packages/tests/tests/peer_exchange.node.spec.ts

@@ -1 +1,2 @@
+export * from "./is_defined.js";
 export * from "./random_subset.js";

packages/utils/src/common/index.ts

@@ -0,0 +1,3 @@
+export function isDefined<T>(value: T | undefined): value is T {
+  return Boolean(value);
+}