chore: refactor peer-exchange according to nwaku 0.15.0 (#1193)

* merge with master: adhere acc to updated API

* bump go-waku to 0.5.2

.github/workflows/ci.yml

@@ -98,7 +98,7 @@ jobs:
   node_with_go_waku:
     runs-on: ubuntu-latest
     env:
-      GO_WAKU_VERSION: "0.5.1"
+      GO_WAKU_VERSION: "0.5.2"
       WAKU_SERVICE_NODE_DIR: ../../go-waku
       WAKU_SERVICE_NODE_BIN: ./waku
       WAKU_SERVICE_NODE_PARAMS: "--min-relay-peers-to-publish=0" # Can be removed once https://github.com/status-im/nwaku/issues/1004 is done

packages/interfaces/src/peer_exchange.ts

@@ -1,16 +1,12 @@
 import type { ConnectionManager } from "@libp2p/interface-connection-manager";
 import type { PeerId } from "@libp2p/interface-peer-id";
 import type { PeerStore } from "@libp2p/interface-peer-store";
-import type { Registrar } from "@libp2p/interface-registrar";
 
 import { IEnr } from "./enr.js";
 import { PointToPointProtocol } from "./protocols.js";
 
 export interface IPeerExchange extends PointToPointProtocol {
-  query(
-    params: PeerExchangeQueryParams,
-    callback: (response: PeerExchangeResponse) => Promise<void> | void
-  ): Promise<void>;
+  query(params: PeerExchangeQueryParams): Promise<PeerInfo[]>;
 }
 
 export interface PeerExchangeQueryParams {
@@ -29,5 +25,4 @@ export interface PeerInfo {
 export interface PeerExchangeComponents {
   connectionManager: ConnectionManager;
   peerStore: PeerStore;
-  registrar: Registrar;
 }

packages/peer-exchange/src/waku_peer_exchange.ts

@@ -1,20 +1,17 @@
 import type { ConnectionManager } from "@libp2p/interface-connection-manager";
 import type { PeerStore } from "@libp2p/interface-peer-store";
-import type {
-  IncomingStreamData,
-  Registrar,
-} from "@libp2p/interface-registrar";
 import { BaseProtocol } from "@waku/core/lib/base_protocol";
 import { EnrDecoder } from "@waku/enr";
 import type {
   IPeerExchange,
   PeerExchangeQueryParams,
-  PeerExchangeResponse,
+  PeerInfo,
 } from "@waku/interfaces";
 import debug from "debug";
 import all from "it-all";
 import * as lp from "it-length-prefixed";
 import { pipe } from "it-pipe";
+import { Uint8ArrayList } from "uint8arraylist";
 
 import { PeerExchangeRPC } from "./rpc.js";
 
@@ -24,7 +21,6 @@ const log = debug("waku:peer-exchange");
 
 export interface PeerExchangeComponents {
   peerStore: PeerStore;
-  registrar: Registrar;
   connectionManager: ConnectionManager;
 }
 
@@ -32,9 +28,7 @@ export interface PeerExchangeComponents {
  * Implementation of the Peer Exchange protocol (https://rfc.vac.dev/spec/34/)
  */
 export class WakuPeerExchange extends BaseProtocol implements IPeerExchange {
-  private callback:
-    | ((response: PeerExchangeResponse) => Promise<void>)
-    | undefined;
+  multicodec: string;
 
   /**
    * @param components - libp2p components
@@ -47,20 +41,13 @@ export class WakuPeerExchange extends BaseProtocol implements IPeerExchange {
         components.connectionManager
       )
     );
-    this.components.registrar
-      .handle(PeerExchangeCodec, this.handler.bind(this))
-      .catch((e) => log("Failed to register peer exchange protocol", e));
+    this.multicodec = PeerExchangeCodec;
   }
 
   /**
    * Make a peer exchange query to a peer
    */
-  async query(
-    params: PeerExchangeQueryParams,
-    callback: (response: PeerExchangeResponse) => Promise<void>
-  ): Promise<void> {
-    this.callback = callback;
-
+  async query(params: PeerExchangeQueryParams): Promise<PeerInfo[]> {
     const { numPeers } = params;
 
     const rpcQuery = PeerExchangeRPC.createRequest({
@@ -71,22 +58,20 @@ export class WakuPeerExchange extends BaseProtocol implements IPeerExchange {
 
     const stream = await this.newStream(peer);
 
-    await pipe(
+    const res = await pipe(
       [rpcQuery.encode()],
       lp.encode(),
       stream,
       lp.decode(),
       async (source) => await all(source)
     );
-  }
 
-  /**
-   * Handle a peer exchange query response
-   */
-  private handler(streamData: IncomingStreamData): void {
-    const { stream } = streamData;
-    pipe(stream, lp.decode(), async (source) => {
-      for await (const bytes of source) {
+    try {
+      const bytes = new Uint8ArrayList();
+      res.forEach((chunk) => {
+        bytes.append(chunk);
+      });
+
       const decoded = PeerExchangeRPC.decode(bytes).response;
 
       if (!decoded) {
@@ -105,11 +90,11 @@ export class WakuPeerExchange extends BaseProtocol implements IPeerExchange {
         };
       });
 
-        if (!this.callback) throw new Error("Callback not set");
-
-        await this.callback({ peerInfos });
+      return peerInfos;
+    } catch (err) {
+      log("Failed to decode push reply", err);
+      throw new Error("Failed to decode push reply");
     }
-    }).catch((err) => log("Failed to handle peer exchange request", err));
   }
 }
 

packages/peer-exchange/src/waku_peer_exchange_discovery.ts

@@ -149,14 +149,11 @@ export class PeerExchangeDiscovery
     }, queryInterval * currentAttempt);
   };
 
-  private query(peerId: PeerId): Promise<void> {
-    return this.peerExchange.query(
-      {
+  private async query(peerId: PeerId): Promise<void> {
+    const peerInfos = await this.peerExchange.query({
       numPeers: DEFAULT_PEER_EXCHANGE_REQUEST_NODES,
       peerId,
-      },
-      async (response) => {
-        const { peerInfos } = response;
+    });
 
     for (const _peerInfo of peerInfos) {
       const { ENR } = _peerInfo;
@@ -165,25 +162,21 @@ export class PeerExchangeDiscovery
         continue;
       }
 
-          const peerInfo = ENR.peerInfo;
+      const { peerId, peerInfo } = ENR;
+
+      if (!peerId || !peerInfo) continue;
+
+      const { multiaddrs } = peerInfo;
 
       if (
-            !peerInfo ||
-            !peerInfo.id ||
-            !peerInfo.multiaddrs ||
-            !peerInfo.multiaddrs.length
-          )
-            continue;
-
-          if (
-            (await this.components.peerStore.getTags(peerInfo.id)).find(
+        (await this.components.peerStore.getTags(peerId)).find(
           ({ name }) => name === DEFAULT_PEER_EXCHANGE_TAG_NAME
         )
       )
         continue;
 
       await this.components.peerStore.tagPeer(
-            peerInfo.id,
+        peerId,
         DEFAULT_PEER_EXCHANGE_TAG_NAME,
         {
           value: this.options.tagValue ?? DEFAULT_PEER_EXCHANGE_TAG_VALUE,
@@ -193,13 +186,15 @@ export class PeerExchangeDiscovery
 
       this.dispatchEvent(
         new CustomEvent<PeerInfo>("peer", {
-              detail: peerInfo,
+          detail: {
+            id: peerId,
+            multiaddrs,
+            protocols: [],
+          },
         })
       );
     }
   }
-    );
-  }
 
   private abortQueriesForPeer(peerIdStr: string): void {
     log(`Aborting queries for peer: ${peerIdStr}`);

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

@@ -4,7 +4,7 @@ import {
   getPredefinedBootstrapNodes,
 } from "@waku/core/lib/predefined_bootstrap_nodes";
 import { createLightNode } from "@waku/create";
-import type { LightNode, PeerExchangeResponse } from "@waku/interfaces";
+import type { LightNode } from "@waku/interfaces";
 import {
   PeerExchangeCodec,
   WakuPeerExchange,
@@ -18,19 +18,17 @@ import { Nwaku } from "../src/nwaku.js";
 describe("Peer Exchange", () => {
   let waku: LightNode;
 
-  before(async function () {
-    // skipping in CI as this test demonstrates Peer Exchange working with the test fleet
-    // but not with locally run nwaku nodes
-    if (process.env.CI) {
-      this.skip();
-    }
-  });
-
   afterEach(async function () {
     !!waku && waku.stop().catch((e) => console.log("Waku failed to stop", e));
   });
 
   it("Auto discovery", async function () {
+    // skipping in CI as this test demonstrates Peer Exchange working with the test fleet
+    // but not with locally run nwaku nodes
+    if (process.env.CI) {
+      this.skip();
+    }
+
     this.timeout(50_000);
 
     waku = await createLightNode({
@@ -77,7 +75,7 @@ describe("Peer Exchange", () => {
     });
 
     it("nwaku interop", async function () {
-      this.timeout(25_000);
+      this.timeout(15_000);
 
       await nwaku1.start({
         discv5Discovery: true,
@@ -115,45 +113,28 @@ describe("Peer Exchange", () => {
         // @ts-ignore
         connectionManager: waku.libp2p.connectionManager,
         peerStore: waku.libp2p.peerStore,
-        // eslint-disable-next-line @typescript-eslint/ban-ts-comment
-        // @ts-ignore
-        registrar: waku.libp2p.registrar,
       });
 
-      let receivedCallback = false;
-
       const numPeersToRequest = 1;
-      const callback = async (
-        response: PeerExchangeResponse
-      ): Promise<void> => {
-        const doesMultiaddrExist = response.peerInfos.find(
+
+      const peerInfos = await peerExchange.query({
+        numPeers: numPeersToRequest,
+      });
+
+      expect(peerInfos.length).to.be.greaterThan(0);
+      expect(peerInfos.length).to.be.lessThanOrEqual(numPeersToRequest);
+      expect(peerInfos[0].ENR).to.not.be.null;
+
+      const doesMultiaddrExist =
+        peerInfos.find(
           (peerInfo) =>
             peerInfo.ENR?.getFullMultiaddrs()?.find((multiaddr) =>
               multiaddr.equals(nwaku1Ma)
             ) !== undefined
-        );
-
-        expect(response.peerInfos.length).to.be.greaterThan(0);
-        expect(response.peerInfos.length).to.be.lessThanOrEqual(
-          numPeersToRequest
-        );
-        expect(response.peerInfos[0].ENR).to.not.be.null;
-
+        ) !== undefined;
       expect(doesMultiaddrExist).to.be.equal(true);
 
       expect(waku.libp2p.peerStore.has(await nwaku2.getPeerId())).to.be.true;
-
-        receivedCallback = true;
-      };
-
-      await peerExchange.query(
-        {
-          numPeers: numPeersToRequest,
-        },
-        callback
-      );
-
-      expect(receivedCallback).to.be.true;
     });
   });
 });