Bump libp2p from 0.32.4 to 0.36.2, libp2p-gossipsub from 0.12.1 to 0.13.0 (#470)

Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Changed
+
+- **Breaking**: Upgrade `libp2p` to `0.36.2` & `libp2p-gossipsub` to `0.13.0`. Some APIs are now async.
+
 ## [0.16.0] - 2022-01-31
 
 ### Changed

examples/eth-pm-wallet-encryption/src/App.tsx

@@ -143,10 +143,16 @@ function App() {
   useEffect(() => {
     if (!waku) return;
 
-    const interval = setInterval(() => {
+    const interval = setInterval(async () => {
+      let lightPushPeers = 0;
+      // eslint-disable-next-line @typescript-eslint/no-unused-vars
+      for await (const _peer of waku.store.peers) {
+        lightPushPeers++;
+      }
+
       setPeerStats({
         relayPeers: waku.relay.getPeers().size,
-        lightPushPeers: waku.lightPush.peers.length,
+        lightPushPeers,
       });
     }, 1000);
     return () => clearInterval(interval);

examples/eth-pm/src/App.tsx

@@ -162,10 +162,16 @@ function App() {
   useEffect(() => {
     if (!waku) return;
 
-    const interval = setInterval(() => {
+    const interval = setInterval(async () => {
+      let lightPushPeers = 0;
+      // eslint-disable-next-line @typescript-eslint/no-unused-vars
+      for await (const _peer of waku.store.peers) {
+        lightPushPeers++;
+      }
+
       setPeerStats({
         relayPeers: waku.relay.getPeers().size,
-        lightPushPeers: waku.lightPush.peers.length,
+        lightPushPeers,
       });
     }, 1000);
     return () => clearInterval(interval);

examples/web-chat/src/App.tsx

@@ -130,7 +130,7 @@ export default function App() {
 
     const retrieveMessages = async () => {
       await waku.waitForRemotePeer();
-      console.log(`Retrieving archived messages}`);
+      console.log(`Retrieving archived messages`);
 
       try {
         retrieveStoreMessages(waku, dispatchMessages).then((length) => {
@@ -156,11 +156,14 @@ export default function App() {
             nick={nick}
             messages={messages}
             commandHandler={(input: string) => {
-              const { command, response } = handleCommand(input, waku, setNick);
-              const commandMessages = response.map((msg) => {
-                return Message.fromUtf8String(command, msg);
-              });
-              dispatchMessages(commandMessages);
+              handleCommand(input, waku, setNick).then(
+                ({ command, response }) => {
+                  const commandMessages = response.map((msg) => {
+                    return Message.fromUtf8String(command, msg);
+                  });
+                  dispatchMessages(commandMessages);
+                }
+              );
             }}
           />
         </ThemeProvider>

examples/web-chat/src/Room.tsx

@@ -6,6 +6,8 @@ import { useWaku } from './WakuContext';
 import { TitleBar } from '@livechat/ui-kit';
 import { Message } from './Message';
 import { ChatMessage } from './chat_message';
+import { useEffect, useState } from 'react';
+import PeerId from 'peer-id';
 
 interface Props {
   messages: Message[];
@@ -16,12 +18,41 @@ interface Props {
 export default function Room(props: Props) {
   const { waku } = useWaku();
 
-  let relayPeers = 0;
-  let storePeers = 0;
-  if (waku) {
-    relayPeers = waku.relay.getPeers().size;
-    storePeers = waku.store.peers.length;
-  }
+  const [peers, setPeers] = useState<PeerId[]>([]);
+  const [storePeers, setStorePeers] = useState(0);
+  const [relayPeers, setRelayPeers] = useState(0);
+
+  useEffect(() => {
+    // Add a peer to the list every time a connection happen to ensure the stats update correctly
+    if (!waku) return;
+
+    const addPeer = (event: { peerId: PeerId }) => {
+      setPeers((peers) => {
+        return [...peers, event.peerId];
+      });
+    };
+
+    waku.libp2p.peerStore.on('change:protocols', addPeer);
+
+    return () => {
+      waku.libp2p.connectionManager.removeListener('change:protocols', addPeer);
+    };
+  }, [waku]);
+
+  useEffect(() => {
+    if (!waku) return;
+
+    setRelayPeers(waku.relay.getPeers().size);
+
+    (async () => {
+      let counter = 0;
+      // eslint-disable-next-line @typescript-eslint/no-unused-vars
+      for await (const _peer of waku.store.peers) {
+        counter++;
+      }
+      setStorePeers(counter);
+    })();
+  }, [waku, peers]);
 
   return (
     <div

examples/web-chat/src/command.ts

@@ -53,13 +53,18 @@ function connect(peer: string | undefined, waku: Waku | undefined): string[] {
   }
 }
 
-function peers(waku: Waku | undefined): string[] {
+async function peers(waku: Waku | undefined): Promise<string[]> {
   if (!waku) {
     return ['Waku node is starting'];
   }
   let response: string[] = [];
-  waku.libp2p.peerStore.peers.forEach((peer, peerId) => {
-    response.push(peerId + ':');
+  const peers = [];
+
+  for await (const peer of waku.libp2p.peerStore.getPeers()) {
+    peers.push(peer);
+  }
+  Array.from(peers).forEach((peer) => {
+    response.push(peer.id.toB58String() + ':');
     let addresses = '  addresses: [';
     peer.addresses.forEach(({ multiaddr }) => {
       addresses += ' ' + multiaddr.toString() + ',';
@@ -104,11 +109,11 @@ function connections(waku: Waku | undefined): string[] {
   return response;
 }
 
-export default function handleCommand(
+export default async function handleCommand(
   input: string,
   waku: Waku | undefined,
   setNick: (nick: string) => void
-): { command: string; response: string[] } {
+): Promise<{ command: string; response: string[] }> {
   let response: string[] = [];
   const args = parseInput(input);
   const command = args.shift()!;
@@ -126,7 +131,7 @@ export default function handleCommand(
       connect(args.shift(), waku).map((str) => response.push(str));
       break;
     case '/peers':
-      peers(waku).map((str) => response.push(str));
+      (await peers(waku)).map((str) => response.push(str));
       break;
     case '/connections':
       connections(waku).map((str) => response.push(str));

package.json

@@ -67,9 +67,9 @@
     "it-concat": "^2.0.0",
     "it-length-prefixed": "^5.0.2",
     "js-sha3": "^0.8.0",
-    "libp2p": "^0.32.4",
+    "libp2p": "^0.36.2",
     "libp2p-bootstrap": "^0.14.0",
-    "libp2p-gossipsub": "^0.12.1",
+    "libp2p-gossipsub": "^0.13.0",
     "libp2p-mplex": "^0.10.4",
     "libp2p-websockets": "^0.16.1",
     "multiaddr": "^10.0.1",

src/lib/select_peer.ts

@@ -5,7 +5,14 @@ import { Peer } from 'libp2p/src/peer-store';
  * Returns a pseudo-random peer that supports the given protocol.
  * Useful for protocols such as store and light push
  */
-export function selectRandomPeer(peers: Peer[]): Peer | undefined {
+export async function selectRandomPeer(
+  peersIter: AsyncIterable<Peer>
+): Promise<Peer | undefined> {
+  const peers = [];
+  for await (const peer of peersIter) {
+    peers.push(peer);
+  }
+
   if (peers.length === 0) return;
 
   const index = Math.round(Math.random() * (peers.length - 1));
@@ -15,8 +22,14 @@ export function selectRandomPeer(peers: Peer[]): Peer | undefined {
 /**
  * Returns the list of peers that supports the given protocol.
  */
-export function getPeersForProtocol(libp2p: Libp2p, protocol: string): Peer[] {
-  return Array.from(libp2p.peerStore.peers.values()).filter((peer) =>
-    peer.protocols.includes(protocol)
-  );
+export async function* getPeersForProtocol(
+  libp2p: Libp2p,
+  protocol: string
+): AsyncIterable<Peer> {
+  for await (const peer of libp2p.peerStore.getPeers()) {
+    if (!peer.protocols.includes(protocol)) {
+      continue;
+    }
+    yield peer;
+  }
 }

src/lib/waku.node.spec.ts

@@ -40,9 +40,7 @@ describe('Waku Dial [node only]', function () {
       await waku.waitForRemotePeer([Protocols.Relay]);
 
       const nimPeerId = await nimWaku.getPeerId();
-      const jsPeers = waku.libp2p.peerStore.peers;
-
-      expect(jsPeers.has(nimPeerId.toB58String())).to.be.true;
+      expect(await waku.libp2p.peerStore.has(nimPeerId)).to.be.true;
     });
   });
 
@@ -245,7 +243,12 @@ describe('Wait for remote peer / get peers', function () {
     });
     await waku.dial(multiAddrWithId);
     await waku.waitForRemotePeer([Protocols.Store]);
-    const peers = waku.store.peers.map((peer) => peer.id.toB58String());
+
+    const peers = [];
+    for await (const peer of waku.store.peers) {
+      peers.push(peer.id.toB58String());
+    }
+
     const nimPeerId = multiAddrWithId.getPeerId();
 
     expect(nimPeerId).to.not.be.undefined;
@@ -263,7 +266,12 @@ describe('Wait for remote peer / get peers', function () {
     });
     await waku.dial(multiAddrWithId);
     await waku.waitForRemotePeer([Protocols.LightPush]);
-    const peers = waku.lightPush.peers.map((peer) => peer.id.toB58String());
+
+    const peers = [];
+    for await (const peer of waku.lightPush.peers) {
+      peers.push(peer.id.toB58String());
+    }
+
     const nimPeerId = multiAddrWithId.getPeerId();
 
     expect(nimPeerId).to.not.be.undefined;

src/lib/waku.ts

@@ -13,7 +13,7 @@ import Websockets from 'libp2p-websockets';
 // eslint-disable-next-line @typescript-eslint/ban-ts-comment
 // @ts-ignore: No types available
 import filters from 'libp2p-websockets/src/filters';
-import Ping from 'libp2p/src/ping';
+import PingService from 'libp2p/src/ping';
 import { Multiaddr, multiaddr } from 'multiaddr';
 import PeerId from 'peer-id';
 
@@ -336,9 +336,14 @@ export class Waku {
     }
 
     if (desiredProtocols.includes(Protocols.Store)) {
-      const peers = this.store.peers;
+      let storePeerFound = false;
 
-      if (peers.length == 0) {
+      for await (const _peer of this.store.peers) {
+        storePeerFound = true;
+        break;
+      }
+
+      if (!storePeerFound) {
         // No peer available for this protocol, waiting to connect to one.
         const promise = new Promise<void>((resolve) => {
           this.libp2p.peerStore.on(
@@ -356,9 +361,14 @@ export class Waku {
     }
 
     if (desiredProtocols.includes(Protocols.LightPush)) {
-      const peers = this.lightPush.peers;
+      let lightPushPeerFound = false;
 
-      if (peers.length == 0) {
+      for await (const _peer of this.lightPush.peers) {
+        lightPushPeerFound = true;
+        break;
+      }
+
+      if (!lightPushPeerFound) {
         // No peer available for this protocol, waiting to connect to one.
         const promise = new Promise<void>((resolve) => {
           this.libp2p.peerStore.on(
@@ -390,8 +400,11 @@ export class Waku {
     const peerIdStr = peerId.toB58String();
 
     if (pingPeriodSecs !== 0) {
+      const pingService = new PingService(this.libp2p);
       this.pingKeepAliveTimers[peerIdStr] = setInterval(() => {
-        Ping(this.libp2p, peerId);
+        pingService.ping(peerId).catch((e) => {
+          dbg(`Ping failed (${peerIdStr})`, e);
+        });
       }, pingPeriodSecs * 1000);
     }
 

src/lib/waku_light_push/index.ts

@@ -52,10 +52,10 @@ export class WakuLightPush {
   ): Promise<PushResponse | null> {
     let peer;
     if (opts?.peerId) {
-      peer = this.libp2p.peerStore.get(opts.peerId);
+      peer = await this.libp2p.peerStore.get(opts.peerId);
       if (!peer) throw 'Peer is unknown';
     } else {
-      peer = this.randomPeer;
+      peer = await this.randomPeer;
     }
     if (!peer) throw 'No peer available';
     if (!peer.protocols.includes(LightPushCodec))
@@ -99,7 +99,7 @@ export class WakuLightPush {
    * Returns known peers from the address book (`libp2p.peerStore`) that support
    * light push protocol. Waku may or  may not be currently connected to these peers.
    */
-  get peers(): Peer[] {
+  get peers(): AsyncIterable<Peer> {
     return getPeersForProtocol(this.libp2p, LightPushCodec);
   }
 
@@ -108,7 +108,7 @@ export class WakuLightPush {
    * book (`libp2p.peerStore`). Waku may or  may not be currently connected to
    * this peer.
    */
-  get randomPeer(): Peer | undefined {
+  get randomPeer(): Promise<Peer | undefined> {
     return selectRandomPeer(this.peers);
   }
 }

src/lib/waku_relay/index.node.spec.ts

@@ -428,12 +428,9 @@ describe('Waku Relay [node only]', () => {
 
         await delay(2000);
         // Check that the two JS peers are NOT directly connected
-        expect(
-          waku1.libp2p.peerStore.peers.has(waku2.libp2p.peerId.toB58String())
-        ).to.be.false;
-        expect(
-          waku2.libp2p.peerStore.peers.has(waku1.libp2p.peerId.toB58String())
-        ).to.be.false;
+        expect(await waku1.libp2p.peerStore.has(waku2.libp2p.peerId)).to.be
+          .false;
+        expect(waku2.libp2p.peerStore.has(waku1.libp2p.peerId)).to.be.false;
 
         const msgStr = 'Hello there!';
         const message = await WakuMessage.fromUtf8String(

src/lib/waku_relay/index.ts

@@ -8,11 +8,7 @@ import {
   PeerScoreParams,
   PeerScoreThresholds,
 } from 'libp2p-gossipsub/src/score';
-import {
-  createGossipRpc,
-  messageIdToString,
-  shuffle,
-} from 'libp2p-gossipsub/src/utils';
+import { createGossipRpc, shuffle } from 'libp2p-gossipsub/src/utils';
 import { InMessage } from 'libp2p-interfaces/src/pubsub';
 import { SignaturePolicy } from 'libp2p-interfaces/src/pubsub/signature-policy';
 import PeerId from 'peer-id';
@@ -112,8 +108,8 @@ export class WakuRelay extends Gossipsub {
    * @override
    * @returns {void}
    */
-  public start(): void {
-    super.start();
+  public async start(): Promise<void> {
+    await super.start();
     this.subscribe(this.pubSubTopic);
   }
 
@@ -331,17 +327,16 @@ export class WakuRelay extends Gossipsub {
    * @returns {void}
    */
   async _publish(msg: InMessage): Promise<void> {
+    const msgIdStr = await this.getCanonicalMsgIdStr(msg);
     if (msg.receivedFrom !== this.peerId.toB58String()) {
-      this.score.deliverMessage(msg);
-      this.gossipTracer.deliverMessage(msg);
+      this.score.deliverMessage(msg, msgIdStr);
+      this.gossipTracer.deliverMessage(msgIdStr);
     }
 
-    const msgID = await this.getMsgId(msg);
-    const msgIdStr = messageIdToString(msgID);
     // put in seen cache
     this.seenCache.put(msgIdStr);
 
-    this.messageCache.put(msg);
+    this.messageCache.put(msg, msgIdStr);
 
     const toSend = new Set<string>();
     msg.topicIDs.forEach((topic) => {
@@ -489,36 +484,47 @@ export class WakuRelay extends Gossipsub {
    * @param {string} id
    * @param {string} topic
    * @param {boolean} doPX
-   * @returns {RPC.IControlPrune}
+   * @returns {Promise<RPC.IControlPrune>}
    */
-  _makePrune(id: string, topic: string, doPX: boolean): RPC.IControlPrune {
+  async _makePrune(
+    id: string,
+    topic: string,
+    doPX: boolean
+  ): Promise<RPC.IControlPrune> {
     // backoff is measured in seconds
     // RelayPruneBackoff is measured in milliseconds
     const backoff = constants.RelayPruneBackoff / 1000;
-    const px: RPC.IPeerInfo[] = [];
-    if (doPX) {
-      // select peers for Peer eXchange
-      const peers = getRelayPeers(
-        this,
-        topic,
-        constants.RelayPrunePeers,
-        (xid: string): boolean => {
-          return xid !== id && this.score.score(xid) >= 0;
-        }
-      );
-      peers.forEach((p) => {
+    if (!doPX) {
+      return {
+        topicID: topic,
+        peers: [],
+        backoff: backoff,
+      };
+    }
+
+    // select peers for Peer eXchange
+    const peers = getRelayPeers(
+      this,
+      topic,
+      constants.RelayPrunePeers,
+      (xid: string): boolean => {
+        return xid !== id && this.score.score(xid) >= 0;
+      }
+    );
+    const px = await Promise.all(
+      Array.from(peers).map(async (p) => {
         // see if we have a signed record to send back; if we don't, just send
         // the peer ID and let the pruned peer find them in the DHT -- we can't trust
         // unsigned address records through PX anyways
         // Finding signed records in the DHT is not supported at the time of writing in js-libp2p
         const peerId = PeerId.createFromB58String(p);
-        px.push({
+        return {
           peerID: peerId.toBytes(),
           signedPeerRecord:
-            this._libp2p.peerStore.addressBook.getRawEnvelope(peerId),
-        });
-      });
-    }
+            await this._libp2p.peerStore.addressBook.getRawEnvelope(peerId),
+        };
+      })
+    );
     return {
       topicID: topic,
       peers: px,

src/lib/waku_store/index.node.spec.ts

@@ -268,10 +268,13 @@ describe('Waku Store', () => {
 
     dbg('Waku nodes connected to nim Waku');
 
-    let lightPushPeers = waku1.lightPush.peers;
-    while (lightPushPeers.length == 0) {
+    let lightPushPeerFound = false;
+    while (!lightPushPeerFound) {
       await delay(100);
-      lightPushPeers = waku1.lightPush.peers;
+      for await (const _peer of waku1.lightPush.peers) {
+        lightPushPeerFound = true;
+        break;
+      }
     }
 
     dbg('Sending messages using light push');
@@ -282,10 +285,13 @@ describe('Waku Store', () => {
       waku1.lightPush.push(clearMessage),
     ]);
 
-    let storePeers = waku2.store.peers;
-    while (storePeers.length == 0) {
+    let storePeerFound = false;
+    while (!storePeerFound) {
       await delay(100);
-      storePeers = waku2.store.peers;
+      for await (const _peer of waku2.store.peers) {
+        storePeerFound = true;
+        break;
+      }
     }
 
     waku2.store.addDecryptionKey(symKey);
@@ -380,10 +386,13 @@ describe('Waku Store', () => {
 
     dbg('Waku nodes connected to nim Waku');
 
-    let lightPushPeers = waku1.lightPush.peers;
-    while (lightPushPeers.length == 0) {
+    let lightPushPeerFound = false;
+    while (!lightPushPeerFound) {
       await delay(100);
-      lightPushPeers = waku1.lightPush.peers;
+      for await (const _peer of waku1.lightPush.peers) {
+        lightPushPeerFound = true;
+        break;
+      }
     }
 
     dbg('Sending messages using light push');
@@ -394,10 +403,13 @@ describe('Waku Store', () => {
       waku1.lightPush.push(clearMessage),
     ]);
 
-    let storePeers = waku2.store.peers;
-    while (storePeers.length == 0) {
+    let storePeerFound = false;
+    while (!storePeerFound) {
       await delay(100);
-      storePeers = waku2.store.peers;
+      for await (const _peer of waku2.store.peers) {
+        storePeerFound = true;
+        break;
+      }
     }
 
     waku2.addDecryptionKey(symKey, {

src/lib/waku_store/index.ts

@@ -150,11 +150,11 @@ export class WakuStore {
 
     let peer;
     if (opts.peerId) {
-      peer = this.libp2p.peerStore.get(opts.peerId);
+      peer = await this.libp2p.peerStore.get(opts.peerId);
       if (!peer)
         throw `Failed to retrieve connection details for provided peer in peer store: ${opts.peerId.toB58String()}`;
     } else {
-      peer = this.randomPeer;
+      peer = await this.randomPeer;
       if (!peer)
         throw 'Failed to find known peer that registers waku store protocol';
     }
@@ -290,7 +290,7 @@ export class WakuStore {
    * Returns known peers from the address book (`libp2p.peerStore`) that support
    * store protocol. Waku may or  may not be currently connected to these peers.
    */
-  get peers(): Peer[] {
+  get peers(): AsyncIterable<Peer> {
     return getPeersForProtocol(this.libp2p, StoreCodec);
   }
 
@@ -299,7 +299,7 @@ export class WakuStore {
    * book (`libp2p.peerStore`). Waku may or  may not be currently connected to
    * this peer.
    */
-  get randomPeer(): Peer | undefined {
+  get randomPeer(): Promise<Peer | undefined> {
     return selectRandomPeer(this.peers);
   }
 }