feat: improve peer manager and re-integrate to light push

.cspell.json

@@ -116,6 +116,7 @@
     "upgrader",
     "vacp",
     "varint",
+    "weboko",
     "waku",
     "wakuconnect",
     "wakunode",

packages/core/package.json

@@ -82,6 +82,7 @@
   },
   "devDependencies": {
     "@libp2p/peer-id": "^5.0.1",
+    "@libp2p/interface": "^2.1.3",
     "@multiformats/multiaddr": "^12.0.0",
     "@rollup/plugin-commonjs": "^25.0.7",
     "@rollup/plugin-json": "^6.0.0",

packages/core/src/index.ts

@@ -15,11 +15,10 @@ export { LightPushCodec, LightPushCore } from "./lib/light_push/index.js";
 export * as waku_store from "./lib/store/index.js";
 export { StoreCore, StoreCodec } from "./lib/store/index.js";
 
-export { ConnectionManager } from "./lib/connection_manager.js";
+export { ConnectionManager } from "./lib/connection_manager/index.js";
 
 export { getHealthManager } from "./lib/health_manager.js";
 
-export { KeepAliveManager } from "./lib/keep_alive_manager.js";
 export { StreamManager } from "./lib/stream_manager/index.js";
 
 export { MetadataCodec, wakuMetadata } from "./lib/metadata/index.js";

packages/core/src/lib/base_protocol.ts

@@ -5,10 +5,7 @@ import type {
   Libp2pComponents,
   PubsubTopic
 } from "@waku/interfaces";
-import { Logger } from "@waku/utils";
-import { getPeersForProtocol, sortPeersByLatency } from "@waku/utils/libp2p";
 
-import { filterPeersByDiscovery } from "./filterPeers.js";
 import { StreamManager } from "./stream_manager/index.js";
 
 /**
@@ -23,7 +20,6 @@ export class BaseProtocol implements IBaseProtocolCore {
   protected constructor(
     public multicodec: string,
     protected components: Libp2pComponents,
-    private log: Logger,
     public readonly pubsubTopics: PubsubTopic[]
   ) {
     this.addLibp2pEventListener = components.events.addEventListener.bind(
@@ -45,73 +41,4 @@ export class BaseProtocol implements IBaseProtocolCore {
   protected async getStream(peer: Peer): Promise<Stream> {
     return this.streamManager.getStream(peer);
   }
-
-  /**
-   * 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.components.peerStore, [this.multicodec]);
-  }
-
-  public async connectedPeers(): Promise<Peer[]> {
-    const peers = await this.allPeers();
-    return peers.filter((peer) => {
-      const connections = this.components.connectionManager.getConnections(
-        peer.id
-      );
-      return connections.length > 0;
-    });
-  }
-
-  /**
-   * Retrieves a list of connected peers that support the protocol. The list is sorted by latency.
-   *
-   * @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. By default, returns all peers available, including bootstrap.
-   */
-  public async getPeers(
-    {
-      numPeers,
-      maxBootstrapPeers
-    }: {
-      numPeers: number;
-      maxBootstrapPeers: number;
-    } = {
-      maxBootstrapPeers: 0,
-      numPeers: 0
-    }
-  ): Promise<Peer[]> {
-    // Retrieve all connected peers that support the protocol & shard (if configured)
-    const allAvailableConnectedPeers = await this.connectedPeers();
-
-    // Filter the peers based on discovery & number of peers requested
-    const filteredPeers = filterPeersByDiscovery(
-      allAvailableConnectedPeers,
-      numPeers,
-      maxBootstrapPeers
-    );
-
-    // Sort the peers by latency
-    const sortedFilteredPeers = await sortPeersByLatency(
-      this.components.peerStore,
-      filteredPeers
-    );
-
-    if (sortedFilteredPeers.length === 0) {
-      this.log.warn(
-        "No peers found. Ensure you have a connection to the network."
-      );
-    }
-
-    if (sortedFilteredPeers.length < numPeers) {
-      this.log.warn(
-        `Only ${sortedFilteredPeers.length} peers found. Requested ${numPeers}.`
-      );
-    }
-
-    return sortedFilteredPeers;
-  }
 }

packages/core/src/lib/connection_manager.ts → packages/core/src/lib/connection_manager/connection_manager.ts

@@ -10,7 +10,6 @@
   IConnectionStateEvents,
   IPeersByDiscoveryEvents,
   IRelay,
-  KeepAliveOptions,
   PeersByDiscoveryResult,
   PubsubTopic,
   ShardInfo
@@ -20,23 +19,36 @@
 import { Logger } from "@waku/utils";
 
 import { KeepAliveManager } from "./keep_alive_manager.js";
+import { getPeerPing } from "./utils.js";
 
 const log = new Logger("connection-manager");
 
-export const DEFAULT_MAX_BOOTSTRAP_PEERS_ALLOWED = 1;
-export const DEFAULT_MAX_DIAL_ATTEMPTS_FOR_PEER = 3;
-export const DEFAULT_MAX_PARALLEL_DIALS = 3;
+const DEFAULT_MAX_BOOTSTRAP_PEERS_ALLOWED = 1;
+const DEFAULT_MAX_DIAL_ATTEMPTS_FOR_PEER = 3;
+const DEFAULT_MAX_PARALLEL_DIALS = 3;
+
+const DEFAULT_PING_KEEP_ALIVE_SEC = 5 * 60;
+const DEFAULT_RELAY_KEEP_ALIVE_SEC = 5 * 60;
+
+type ConnectionManagerConstructorOptions = {
+  libp2p: Libp2p;
+  pubsubTopics: PubsubTopic[];
+  relay?: IRelay;
+  config?: Partial<ConnectionManagerOptions>;
+};
 
 export class ConnectionManager
   extends TypedEventEmitter<IPeersByDiscoveryEvents & IConnectionStateEvents>
   implements IConnectionManager
 {
-  private static instances = new Map<string, ConnectionManager>();
+  // TODO(weboko): make it private
+  public readonly pubsubTopics: PubsubTopic[];
+
   private keepAliveManager: KeepAliveManager;
   private options: ConnectionManagerOptions;
   private libp2p: Libp2p;
   private dialAttemptsForPeer: Map<string, number> = new Map();
   private dialErrorsForPeer: Map<string, any> = new Map();
 
   private currentActiveParallelDialCount = 0;
   private pendingPeerDialQueue: Array<PeerId> = [];
@@ -51,29 +63,6 @@
     return this.isP2PNetworkConnected;
   }
 
-  public static create(
-    peerId: string,
-    libp2p: Libp2p,
-    keepAliveOptions: KeepAliveOptions,
-    pubsubTopics: PubsubTopic[],
-    relay?: IRelay,
-    options?: ConnectionManagerOptions
-  ): ConnectionManager {
-    let instance = ConnectionManager.instances.get(peerId);
-    if (!instance) {
-      instance = new ConnectionManager(
-        libp2p,
-        keepAliveOptions,
-        pubsubTopics,
-        relay,
-        options
-      );
-      ConnectionManager.instances.set(peerId, instance);
-    }
-
-    return instance;
-  }
-
   public stop(): void {
     this.keepAliveManager.stopAll();
     this.libp2p.removeEventListener(
@@ -156,27 +145,26 @@
     };
   }
 
-  private constructor(
-    libp2p: Libp2p,
-    keepAliveOptions: KeepAliveOptions,
-    public readonly configuredPubsubTopics: PubsubTopic[],
-    relay?: IRelay,
-    options?: Partial<ConnectionManagerOptions>
-  ) {
+  public constructor(options: ConnectionManagerConstructorOptions) {
     super();
-    this.libp2p = libp2p;
-    this.configuredPubsubTopics = configuredPubsubTopics;
+    this.libp2p = options.libp2p;
+    this.pubsubTopics = options.pubsubTopics;
     this.options = {
       maxDialAttemptsForPeer: DEFAULT_MAX_DIAL_ATTEMPTS_FOR_PEER,
       maxBootstrapPeersAllowed: DEFAULT_MAX_BOOTSTRAP_PEERS_ALLOWED,
       maxParallelDials: DEFAULT_MAX_PARALLEL_DIALS,
-      ...options
+      pingKeepAlive: DEFAULT_PING_KEEP_ALIVE_SEC,
+      relayKeepAlive: DEFAULT_RELAY_KEEP_ALIVE_SEC,
+      ...options.config
     };
 
     this.keepAliveManager = new KeepAliveManager({
-      relay,
-      libp2p,
-      options: keepAliveOptions
+      relay: options.relay,
+      libp2p: options.libp2p,
+      options: {
+        pingKeepAlive: this.options.pingKeepAlive,
+        relayKeepAlive: this.options.relayKeepAlive
+      }
     });
 
     this.startEventListeners()
@@ -193,6 +181,29 @@
     );
   }
 
+  public async getConnectedPeers(codec?: string): Promise<Peer[]> {
+    const peerIDs = this.libp2p.getPeers();
+
+    if (peerIDs.length === 0) {
+      return [];
+    }
+
+    const peers = await Promise.all(
+      peerIDs.map(async (id) => {
+        try {
+          return await this.libp2p.peerStore.get(id);
+        } catch (e) {
+          return null;
+        }
+      })
+    );
+
+    return peers
+      .filter((p) => !!p)
+      .filter((p) => (codec ? (p as Peer).protocols.includes(codec) : true))
+      .sort((left, right) => getPeerPing(left) - getPeerPing(right)) as Peer[];
+  }
+
   private async dialPeerStorePeers(): Promise<void> {
     const peerInfos = await this.libp2p.peerStore.all();
     const dialPromises = [];
@@ -253,7 +264,7 @@
           // Handle generic error
           log.error(
             `Error dialing peer ${peerId.toString()} - ${
               (error as any).message
             }`
           );
         }
@@ -478,7 +489,7 @@
 
       log.warn(
         `Discovered peer ${peerId.toString()} with ShardInfo ${shardInfo} is not part of any of the configured pubsub topics (${
-          this.configuredPubsubTopics
+          this.pubsubTopics
         }).
             Not dialing.`
       );
@@ -573,7 +584,7 @@
     const pubsubTopics = shardInfoToPubsubTopics(shardInfo);
 
     const isTopicConfigured = pubsubTopics.some((topic) =>
-      this.configuredPubsubTopics.includes(topic)
+      this.pubsubTopics.includes(topic)
     );
     return isTopicConfigured;
   }

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

@@ -0,0 +1 @@
+export { ConnectionManager } from "./connection_manager.js";

packages/core/src/lib/keep_alive_manager.ts → packages/core/src/lib/connection_manager/keep_alive_manager.ts

@@ -1,14 +1,18 @@
 import type { PeerId } from "@libp2p/interface";
 import type { IRelay, Libp2p, PeerIdStr } from "@waku/interfaces";
-import type { KeepAliveOptions } from "@waku/interfaces";
 import { Logger, pubsubTopicToSingleShardInfo } from "@waku/utils";
 import { utf8ToBytes } from "@waku/utils/bytes";
 
-import { createEncoder } from "./message/version_0.js";
+import { createEncoder } from "../message/version_0.js";
 
-export const RelayPingContentTopic = "/relay-ping/1/ping/null";
+const RelayPingContentTopic = "/relay-ping/1/ping/null";
 const log = new Logger("keep-alive");
 
+type KeepAliveOptions = {
+  pingKeepAlive: number;
+  relayKeepAlive: number;
+};
+
 type CreateKeepAliveManagerOptions = {
   options: KeepAliveOptions;
   libp2p: Libp2p;

packages/core/src/lib/connection_manager/utils.ts

@@ -0,0 +1,25 @@
+import type { Peer } from "@libp2p/interface";
+import { bytesToUtf8 } from "@waku/utils/bytes";
+
+/**
+ * Reads peer's metadata and retrieves ping value.
+ * @param peer Peer or null
+ * @returns -1 if no ping attached, otherwise returns ping value
+ */
+export const getPeerPing = (peer: Peer | null): number => {
+  if (!peer) {
+    return -1;
+  }
+
+  try {
+    const bytes = peer.metadata.get("ping");
+
+    if (!bytes) {
+      return -1;
+    }
+
+    return Number(bytesToUtf8(bytes));
+  } catch (e) {
+    return -1;
+  }
+};

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

@@ -40,7 +40,7 @@ export class FilterCore extends BaseProtocol implements IBaseProtocolCore {
     public readonly pubsubTopics: PubsubTopic[],
     libp2p: Libp2p
   ) {
-    super(FilterCodecs.SUBSCRIBE, libp2p.components, log, pubsubTopics);
+    super(FilterCodecs.SUBSCRIBE, libp2p.components, pubsubTopics);
 
     libp2p
       .handle(FilterCodecs.PUSH, this.onRequest.bind(this), {