feat: node and protocols health

packages/core/src/index.ts

@@ -21,6 +21,8 @@ export { waitForRemotePeer } from "./lib/wait_for_remote_peer.js";
 
 export { ConnectionManager } from "./lib/connection_manager.js";
 
+export { getHealthManager } from "./lib/health_manager.js";
+
 export { KeepAliveManager } from "./lib/keep_alive_manager.js";
 export { StreamManager } from "./lib/stream_manager/index.js";
 

packages/core/src/lib/health_manager.ts

@@ -0,0 +1,90 @@
+import {
+ HealthStatus,
+ type IHealthManager,
+ NodeHealth,
+ type ProtocolHealth,
+ Protocols
+} from "@waku/interfaces";
+
+class HealthManager implements IHealthManager {
+ public static instance: HealthManager;
+ private readonly health: NodeHealth;
+
+ private constructor() {
+ this.health = {
+ overallStatus: HealthStatus.Unhealthy,
+ protocolStatuses: new Map()
+ };
+ }
+
+ public static getInstance(): HealthManager {
+ if (!HealthManager.instance) {
+ HealthManager.instance = new HealthManager();
+ }
+ return HealthManager.instance;
+ }
+
+ public getHealthStatus(): HealthStatus {
+ return this.health.overallStatus;
+ }
+
+ public getProtocolStatus(protocol: Protocols): ProtocolHealth | undefined {
+ return this.health.protocolStatuses.get(protocol);
+ }
+
+ public updateProtocolHealth(
+ multicodec: string,
+ connectedPeers: number
+ ): void {
+ const protocol = this.getNameFromMulticodec(multicodec);
+
+ let status: HealthStatus = HealthStatus.Unhealthy;
+ if (connectedPeers == 1) {
+ status = HealthStatus.MinimallyHealthy;
+ } else if (connectedPeers >= 2) {
+ status = HealthStatus.SufficientlyHealthy;
+ }
+
+ this.health.protocolStatuses.set(protocol, {
+ name: protocol,
+ status: status,
+ lastUpdate: new Date()
+ });
+
+ this.updateOverallHealth();
+ }
+
+ private getNameFromMulticodec(multicodec: string): Protocols {
+ let name: Protocols;
+ if (multicodec.includes("filter")) {
+ name = Protocols.Filter;
+ } else if (multicodec.includes("lightpush")) {
+ name = Protocols.LightPush;
+ } else if (multicodec.includes("store")) {
+ name = Protocols.Store;
+ } else {
+ throw new Error(`Unknown protocol: ${multicodec}`);
+ }
+ return name;
+ }
+
+ private updateOverallHealth(): void {
+ const relevantProtocols = [Protocols.LightPush, Protocols.Filter];
+ const statuses = relevantProtocols.map(
+ (p) => this.getProtocolStatus(p)?.status
+ );
+
+ if (statuses.some((status) => status === HealthStatus.Unhealthy)) {
+ this.health.overallStatus = HealthStatus.Unhealthy;
+ } else if (
+ statuses.some((status) => status === HealthStatus.MinimallyHealthy)
+ ) {
+ this.health.overallStatus = HealthStatus.MinimallyHealthy;
+ } else {
+ this.health.overallStatus = HealthStatus.SufficientlyHealthy;
+ }
+ }
+}
+
+export const getHealthManager = (): HealthManager =>
+ HealthManager.getInstance();

packages/interfaces/src/health_manager.ts

@@ -0,0 +1,26 @@
+import { Protocols } from "./protocols";
+
+export enum HealthStatus {
+ Unhealthy = "Unhealthy",
+ MinimallyHealthy = "MinimallyHealthy",
+ SufficientlyHealthy = "SufficientlyHealthy"
+}
+
+export interface IHealthManager {
+ getHealthStatus: () => HealthStatus;
+ getProtocolStatus: (protocol: Protocols) => ProtocolHealth | undefined;
+ updateProtocolHealth: (multicodec: string, connectedPeers: number) => void;
+}
+
+export type NodeHealth = {
+ overallStatus: HealthStatus;
+ protocolStatuses: ProtocolsHealthStatus;
+};
+
+export type ProtocolHealth = {
+ name: Protocols;
+ status: HealthStatus;
+ lastUpdate: Date;
+};
+
+export type ProtocolsHealthStatus = Map<Protocols, ProtocolHealth>;

packages/interfaces/src/index.ts

@@ -17,3 +17,4 @@ export * from "./dns_discovery.js";
 export * from "./metadata.js";
 export * from "./constants.js";
 export * from "./local_storage.js";
+export * from "./health_manager.js";

packages/interfaces/src/protocols.ts

@@ -25,8 +25,8 @@ export type IBaseProtocolCore = {
 };
 
 export type IBaseProtocolSDK = {
- renewPeer: (peerToDisconnect: PeerId) => Promise<Peer>;
  readonly connectedPeers: Peer[];
+ renewPeer: (peerToDisconnect: PeerId) => Promise<Peer>;
  readonly numPeersToUse: number;
 };
 

packages/interfaces/src/waku.ts

@@ -3,6 +3,7 @@ import type { MultiaddrInput } from "@multiformats/multiaddr";
 
 import { IConnectionManager } from "./connection_manager.js";
 import type { IFilterSDK } from "./filter.js";
+import { IHealthManager } from "./health_manager.js";
 import type { Libp2p } from "./libp2p.js";
 import type { ILightPushSDK } from "./light_push.js";
 import { Protocols } from "./protocols.js";
@@ -27,6 +28,8 @@ export interface Waku {
  isStarted(): boolean;
 
  isConnected(): boolean;
+
+ health: IHealthManager;
 }
 
 export interface LightNode extends Waku {

packages/sdk/src/protocols/base_protocol.ts

@@ -1,7 +1,11 @@
 import type { Peer, PeerId } from "@libp2p/interface";
-import { ConnectionManager } from "@waku/core";
+import { ConnectionManager, getHealthManager } from "@waku/core";
 import { BaseProtocol } from "@waku/core/lib/base_protocol";
-import { IBaseProtocolSDK, ProtocolUseOptions } from "@waku/interfaces";
+import {
+ IBaseProtocolSDK,
+ IHealthManager,
+ ProtocolUseOptions
+} from "@waku/interfaces";
 import { delay, Logger } from "@waku/utils";
 
 interface Options {
@@ -14,6 +18,7 @@ const DEFAULT_NUM_PEERS_TO_USE = 3;
 const DEFAULT_MAINTAIN_PEERS_INTERVAL = 30_000;
 
 export class BaseProtocolSDK implements IBaseProtocolSDK {
+ private healthManager: IHealthManager;
  public readonly numPeersToUse: number;
  private peers: Peer[] = [];
  private maintainPeersIntervalId: ReturnType<
@@ -32,6 +37,9 @@ export class BaseProtocolSDK implements IBaseProtocolSDK {
  options: Options
  ) {
  this.log = new Logger(`sdk:${core.multicodec}`);
+
+ this.healthManager = getHealthManager();
+
  this.numPeersToUse = options?.numPeersToUse ?? DEFAULT_NUM_PEERS_TO_USE;
  const maintainPeersInterval =
  options?.maintainPeersInterval ?? DEFAULT_MAINTAIN_PEERS_INTERVAL;
@@ -60,7 +68,11 @@ export class BaseProtocolSDK implements IBaseProtocolSDK {
  );
  }
 
- this.peers = this.peers.filter((peer) => !peer.id.equals(peerToDisconnect));
+ const updatedPeers = this.peers.filter(
+ (peer) => !peer.id.equals(peerToDisconnect)
+ );
+ this.updatePeers(updatedPeers);
+
  this.log.info(
  `Peer ${peerToDisconnect} disconnected and removed from the peer list`
  );
@@ -192,7 +204,9 @@ export class BaseProtocolSDK implements IBaseProtocolSDK {
 
  await Promise.all(dials);
 
- this.peers = [...this.peers, ...additionalPeers];
+ const updatedPeers = [...this.peers, ...additionalPeers];
+ this.updatePeers(updatedPeers);
+
  this.log.info(
  `Added ${additionalPeers.length} new peers, total peers: ${this.peers.length}`
  );
@@ -232,6 +246,14 @@ export class BaseProtocolSDK implements IBaseProtocolSDK {
  throw error;
  }
  }
+
+ private updatePeers(peers: Peer[]): void {
+ this.peers = peers;
+ this.healthManager.updateProtocolHealth(
+ this.core.multicodec,
+ this.peers.length
+ );
+ }
 }
 
 class RenewPeerLocker {

packages/sdk/src/waku.ts

@@ -1,9 +1,10 @@
 import type { Stream } from "@libp2p/interface";
 import { isPeerId, PeerId } from "@libp2p/interface";
 import { multiaddr, Multiaddr, MultiaddrInput } from "@multiformats/multiaddr";
-import { ConnectionManager } from "@waku/core";
+import { ConnectionManager, getHealthManager } from "@waku/core";
 import type {
  IFilterSDK,
+ IHealthManager,
  ILightPushSDK,
  IRelay,
  IStoreSDK,
@@ -68,6 +69,7 @@ export class WakuNode implements Waku {
  public lightPush?: ILightPushSDK;
  public connectionManager: ConnectionManager;
  public readonly pubsubTopics: PubsubTopic[];
+ public readonly health: IHealthManager;
 
  public constructor(
  options: WakuOptions,
@@ -105,6 +107,8 @@ export class WakuNode implements Waku {
  this.relay
  );
 
+ this.health = getHealthManager();
+
  if (protocolsEnabled.store) {
  const store = wakuStore(this.connectionManager, options);
  this.store = store(libp2p);

packages/tests/src/utils/index.ts

@@ -7,3 +7,4 @@ export * from "./base64_utf8.js";
 export * from "./waitForConnections.js";
 export * from "./custom_mocha_hooks.js";
 export * from "./waku_versions_utils.js";
+export * from "./nodes.js";

packages/tests/src/utils/nodes.ts

@@ -0,0 +1,115 @@
+import { waitForRemotePeer } from "@waku/core";
+import {
+ LightNode,
+ ProtocolCreateOptions,
+ Protocols,
+ ShardingParams,
+ Waku
+} from "@waku/interfaces";
+import { createLightNode } from "@waku/sdk";
+import { isDefined, shardInfoToPubsubTopics } from "@waku/utils";
+import { Context } from "mocha";
+import pRetry from "p-retry";
+
+import { DefaultTestPubsubTopic, NOISE_KEY_1 } from "../constants";
+import { ServiceNodesFleet } from "../lib";
+import { Args } from "../types";
+
+import { waitForConnections } from "./waitForConnections";
+
+export async function runMultipleNodes(
+ context: Context,
+ shardInfo?: ShardingParams,
+ customArgs?: Args,
+ strictChecking: boolean = false,
+ numServiceNodes = 3,
+ withoutFilter = false
+): Promise<[ServiceNodesFleet, LightNode]> {
+ const pubsubTopics = shardInfo
+ ? shardInfoToPubsubTopics(shardInfo)
+ : [DefaultTestPubsubTopic];
+ // create numServiceNodes nodes
+ const serviceNodes = await ServiceNodesFleet.createAndRun(
+ context,
+ pubsubTopics,
+ numServiceNodes,
+ strictChecking,
+ shardInfo,
+ customArgs,
+ withoutFilter
+ );
+
+ const wakuOptions: ProtocolCreateOptions = {
+ staticNoiseKey: NOISE_KEY_1,
+ libp2p: {
+ addresses: { listen: ["/ip4/0.0.0.0/tcp/0/ws"] }
+ }
+ };
+
+ if (shardInfo) {
+ wakuOptions.shardInfo = shardInfo;
+ } else {
+ wakuOptions.pubsubTopics = pubsubTopics;
+ }
+
+ const waku = await createLightNode(wakuOptions);
+ await waku.start();
+
+ if (!waku) {
+ throw new Error("Failed to initialize waku");
+ }
+
+ for (const node of serviceNodes.nodes) {
+ await waku.dial(await node.getMultiaddrWithId());
+ await waitForRemotePeer(
+ waku,
+ [
+ !customArgs?.filter ? undefined : Protocols.Filter,
+ !customArgs?.lightpush ? undefined : Protocols.LightPush
+ ].filter(isDefined)
+ );
+ await node.ensureSubscriptions(pubsubTopics);
+
+ const wakuConnections = waku.libp2p.getConnections();
+ const nodePeers = await node.peers();
+
+ if (wakuConnections.length < 1 || nodePeers.length < 1) {
+ throw new Error(
+ `Expected at least 1 peer in each node. Got waku connections: ${wakuConnections.length} and service nodes: ${nodePeers.length}`
+ );
+ }
+ }
+
+ await waitForConnections(numServiceNodes, waku);
+
+ return [serviceNodes, waku];
+}
+
+export async function teardownNodesWithRedundancy(
+ serviceNodes: ServiceNodesFleet,
+ wakuNodes: Waku | Waku[]
+): Promise<void> {
+ const wNodes = Array.isArray(wakuNodes) ? wakuNodes : [wakuNodes];
+
+ const stopNwakuNodes = serviceNodes.nodes.map(async (node) => {
+ await pRetry(
+ async () => {
+ await node.stop();
+ },
+ { retries: 3 }
+ );
+ });
+
+ const stopWakuNodes = wNodes.map(async (waku) => {
+ if (waku) {
+ await pRetry(
+ async () => {
+ await waku.stop();
+ },
+ { retries: 3 }
+ );
+ }
+ });
+
+ await Promise.all([...stopNwakuNodes, ...stopWakuNodes]);
+}