chore: use peer metadata to gauge protocol usage

.github/workflows/playwright.yml

@@ -20,7 +20,7 @@ jobs:
     timeout-minutes: 60
     runs-on: ubuntu-latest
     container:
-      image: mcr.microsoft.com/playwright:v1.46.0-jammy 
+      image: mcr.microsoft.com/playwright:v1.48.0-jammy
     steps:
     - uses: actions/checkout@v3
     - uses: actions/setup-node@v3

packages/core/src/lib/base_protocol.ts

@@ -1,16 +1,12 @@
 import type { Libp2p } from "@libp2p/interface";
-import type { Peer, PeerStore, Stream } from "@libp2p/interface";
+import type { Peer, Stream } from "@libp2p/interface";
 import type {
   IBaseProtocolCore,
   Libp2pComponents,
   PubsubTopic
 } from "@waku/interfaces";
-import { Logger, pubsubTopicsToShardInfo } from "@waku/utils";
-import {
-  getConnectedPeersForProtocolAndShard,
-  getPeersForProtocol,
-  sortPeersByLatency
-} from "@waku/utils/libp2p";
+import { Logger } from "@waku/utils";
+import { getPeersForProtocol, sortPeersByLatency } from "@waku/utils/libp2p";
 
 import { filterPeersByDiscovery } from "./filterPeers.js";
 import { StreamManager } from "./stream_manager/index.js";
@@ -26,7 +22,7 @@ export class BaseProtocol implements IBaseProtocolCore {
 
   protected constructor(
     public multicodec: string,
-    private components: Libp2pComponents,
+    protected components: Libp2pComponents,
     private log: Logger,
     public readonly pubsubTopics: PubsubTopic[]
   ) {
@@ -50,25 +46,22 @@ export class BaseProtocol implements IBaseProtocolCore {
     return this.streamManager.getStream(peer);
   }
 
-  public get peerStore(): PeerStore {
-    return this.components.peerStore;
-  }
-
   /**
    * Returns known peers from the address book (`libp2p.peerStore`) that support
    * the class protocol. Waku may or may not be currently connected to these
    * peers.
    */
   public async allPeers(): Promise<Peer[]> {
-    return getPeersForProtocol(this.peerStore, [this.multicodec]);
+    return getPeersForProtocol(this.components.peerStore, [this.multicodec]);
   }
 
   public async connectedPeers(): Promise<Peer[]> {
     const peers = await this.allPeers();
     return peers.filter((peer) => {
-      return (
-        this.components.connectionManager.getConnections(peer.id).length > 0
+      const connections = this.components.connectionManager.getConnections(
+        peer.id
       );
+      return connections.length > 0;
     });
   }
 
@@ -77,9 +70,8 @@ export class BaseProtocol implements IBaseProtocolCore {
    *
    * @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 list of peers that support the protocol sorted by latency.
-  */
+   * @returns A list of peers that support the protocol sorted by latency. By default, returns all peers available, including bootstrap.
+   */
   public async getPeers(
     {
       numPeers,
@@ -88,29 +80,23 @@ export class BaseProtocol implements IBaseProtocolCore {
       numPeers: number;
       maxBootstrapPeers: number;
     } = {
-      maxBootstrapPeers: 1,
+      maxBootstrapPeers: 0,
       numPeers: 0
     }
   ): Promise<Peer[]> {
     // Retrieve all connected peers that support the protocol & shard (if configured)
-    const connectedPeersForProtocolAndShard =
-      await getConnectedPeersForProtocolAndShard(
-        this.components.connectionManager.getConnections(),
-        this.peerStore,
-        [this.multicodec],
-        pubsubTopicsToShardInfo(this.pubsubTopics)
-      );
+    const allAvailableConnectedPeers = await this.connectedPeers();
 
     // Filter the peers based on discovery & number of peers requested
     const filteredPeers = filterPeersByDiscovery(
-      connectedPeersForProtocolAndShard,
+      allAvailableConnectedPeers,
       numPeers,
       maxBootstrapPeers
     );
 
     // Sort the peers by latency
     const sortedFilteredPeers = await sortPeersByLatency(
-      this.peerStore,
+      this.components.peerStore,
       filteredPeers
     );
 

packages/core/src/lib/connection_manager.ts

@@ -34,9 +34,9 @@
   private static instances = new Map<string, ConnectionManager>();
   private keepAliveManager: KeepAliveManager;
   private options: ConnectionManagerOptions;
-  private libp2p: Libp2p;
+  public libp2p: Libp2p;
   private dialAttemptsForPeer: Map<string, number> = new Map();
   private dialErrorsForPeer: Map<string, any> = new Map();
 
   private currentActiveParallelDialCount = 0;
   private pendingPeerDialQueue: Array<PeerId> = [];
@@ -253,7 +253,7 @@
           // Handle generic error
           log.error(
             `Error dialing peer ${peerId.toString()} - ${
               (error as any).message
             }`
           );
         }

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

@@ -301,8 +301,17 @@ export class FilterCore extends BaseProtocol implements IBaseProtocolCore {
         () => {
           log.info("Receiving pipe closed.");
         },
-        (e) => {
-          log.error("Error with receiving pipe", e);
+        async (e) => {
+          log.error(
+            "Error with receiving pipe",
+            e,
+            " -- ",
+            "on peer ",
+            connection.remotePeer.toString(),
+            " -- ",
+            "stream ",
+            stream
+          );
         }
       );
     } catch (e) {

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

@@ -45,7 +45,7 @@ class Metadata extends BaseProtocol implements IMetadata {
       pubsubTopicsToShardInfo(this.pubsubTopics)
     );
 
-    const peer = await this.peerStore.get(peerId);
+    const peer = await this.libp2pComponents.peerStore.get(peerId);
     if (!peer) {
       return {
         shardInfo: null,

packages/discovery/src/peer-exchange/waku_peer_exchange.ts

@@ -47,7 +47,7 @@ export class WakuPeerExchange extends BaseProtocol implements IPeerExchange {
       numPeers: BigInt(numPeers)
     });
 
-    const peer = await this.peerStore.get(peerId);
+    const peer = await this.components.peerStore.get(peerId);
     if (!peer) {
       return {
         peerInfos: null,

packages/interfaces/src/connection_manager.ts

@@ -1,4 +1,9 @@
-import type { Peer, PeerId, TypedEventEmitter } from "@libp2p/interface";
+import type {
+  Libp2p,
+  Peer,
+  PeerId,
+  TypedEventEmitter
+} from "@libp2p/interface";
 
 import { PubsubTopic } from "./misc.js";
 
@@ -67,4 +72,5 @@ export interface IConnectionManager
   dropConnection(peerId: PeerId): Promise<void>;
   getPeersByDiscovery(): Promise<PeersByDiscoveryResult>;
   stop(): void;
+  libp2p: Libp2p;
 }

packages/interfaces/src/protocols.ts

@@ -1,6 +1,6 @@
 import type { Libp2p } from "@libp2p/interface";
 import type { PeerId } from "@libp2p/interface";
-import type { Peer, PeerStore } from "@libp2p/interface";
+import type { Peer } from "@libp2p/interface";
 
 import type { CreateLibp2pOptions } from "./libp2p.js";
 import type { IDecodedMessage } from "./message.js";
@@ -16,7 +16,6 @@ export enum Protocols {
 
 export type IBaseProtocolCore = {
   multicodec: string;
-  peerStore: PeerStore;
   allPeers: () => Promise<Peer[]>;
   connectedPeers: () => Promise<Peer[]>;
   addLibp2pEventListener: Libp2p["addEventListener"];
@@ -25,7 +24,7 @@ export type IBaseProtocolCore = {
 
 export type IBaseProtocolSDK = {
   readonly connectedPeers: Peer[];
-  renewPeer: (peerToDisconnect: PeerId) => Promise<Peer>;
+  renewPeer: (peerToDisconnect: PeerId) => Promise<Peer | undefined>;
   readonly numPeersToUse: number;
 };
 
@@ -36,10 +35,6 @@ export type NetworkConfig = StaticSharding | AutoSharding;
  * Options for using LightPush and Filter
  */
 export type ProtocolUseOptions = {
-  /**
-   * Optional flag to enable auto-retry with exponential backoff
-   */
-  autoRetry?: boolean;
   /**
    * Optional flag to force using all available peers
    */
@@ -48,14 +43,6 @@ export type ProtocolUseOptions = {
    * Optional maximum number of attempts for exponential backoff
    */
   maxAttempts?: number;
-  /**
-   * Optional initial delay in milliseconds for exponential backoff
-   */
-  initialDelay?: number;
-  /**
-   * Optional maximum delay in milliseconds for exponential backoff
-   */
-  maxDelay?: number;
 };
 
 export type ProtocolCreateOptions = {

packages/sdk/.mocharc.cjs

@@ -1,26 +1,27 @@
 const config = {
-  extension: ['ts'],
-  spec: 'src/**/*.spec.ts',
-  require: ['ts-node/register', 'isomorphic-fetch'],
-  loader: 'ts-node/esm',
-  nodeOptions: [
-    'experimental-specifier-resolution=node',
-    'loader=ts-node/esm'
-  ],
-  exit: true
-};
-
-if (process.env.CI) {
-  console.log("Running tests in parallel");
-  config.parallel = true;
-  config.jobs = 6;
-  console.log("Activating allure reporting");
-  config.reporter = 'mocha-multi-reporters';
-  config.reporterOptions = {
-    configFile: '.mocha.reporters.json'
+    extension: ['ts'],
+    spec: 'src/**/*.spec.ts',
+    require: ['ts-node/register', 'isomorphic-fetch'],
+    loader: 'ts-node/esm',
+    nodeOptions: [
+      'experimental-specifier-resolution=node',
+      'loader=ts-node/esm'
+    ],
+    exit: true
   };
-} else {
-  console.log("Running tests serially. To enable parallel execution update mocha config");
-}
-
-module.exports = config;
+
+  if (process.env.CI) {
+    console.log("Running tests in parallel");
+    config.parallel = true;
+    config.jobs = 6;
+    console.log("Activating allure reporting");
+    config.reporter = 'mocha-multi-reporters';
+    config.reporterOptions = {
+      configFile: '.mocha.reporters.json'
+    };
+  } else {
+    console.log("Running tests serially. To enable parallel execution update mocha config");
+  }
+
+  module.exports = config;
+

packages/sdk/package.json

@@ -61,7 +61,7 @@
   },
   "dependencies": {
     "@chainsafe/libp2p-noise": "^15.1.0",
-    "@libp2p/bootstrap": "^10.1.2",
+    "@libp2p/bootstrap": "^10",
     "@libp2p/identify": "^2.1.2",
     "@libp2p/mplex": "^10.1.2",
     "@libp2p/ping": "^1.1.2",
@@ -73,23 +73,26 @@
     "@waku/proto": "^0.0.8",
     "@waku/utils": "0.0.20",
     "@waku/message-hash": "0.1.16",
-    "libp2p": "^1.8.1"
+    "async-mutex": "^0.5.0",
+    "libp2p": "^1.8.1",
+    "p-event": "^6.0.1"
   },
   "devDependencies": {
-    "@types/mocha": "^10.0.6",
     "@types/chai": "^4.3.11",
     "@rollup/plugin-commonjs": "^25.0.7",
     "@rollup/plugin-json": "^6.0.0",
     "@rollup/plugin-node-resolve": "^15.2.3",
     "@rollup/plugin-replace": "^5.0.5",
+    "@types/mocha": "^10.0.9",
     "@waku/build-utils": "*",
-    "mocha": "^10.3.0",
-    "sinon": "^18.0.0",
-    "chai": "^4.3.10",
+    "chai": "^5.1.1",
     "cspell": "^8.6.1",
     "interface-datastore": "^8.2.10",
+    "mocha": "^10.7.3",
     "npm-run-all": "^4.1.5",
-    "rollup": "^4.12.0"
+    "rollup": "^4.12.0",
+    "sinon": "^19.0.2",
+    "ts-node": "^10.9.2"
   },
   "peerDependencies": {
     "@libp2p/bootstrap": "^10"