feat: lightpush & filter send requests to multiple peers (#1779)

* feat: lightpush & filter send requests to multiple peers

* build message-hash before core

* chore: restructure folder heirrarchy

* fix: imports

* chore: move @waku/core to dev deps

* feat: create a new `ServiceNodes` wrapper class to encapsulate service node and message collector with redundancy accounted for

* chore(filter): move tests against single service node to a subdir

* feat: support relay, add strict checking, add tests

* fix(filter): handle errors

* chore(tests): add tests for ping

* add tests for push

* chore: abstract redundancy

* feat: add unsubscribe tests

* fix: tests

* add lightpush tests

* fix: imports

* fix: merge & add warning

* merge: master

* fix: breaking tests with master

* address comments

* make num peers configurable

* fix: typo

package.json

@@ -5,11 +5,11 @@
   "workspaces": [
     "packages/interfaces",
     "packages/utils",
+    "packages/message-hash",
     "packages/proto",
     "packages/enr",
     "packages/core",
     "packages/relay",
-    "packages/message-hash",
     "packages/peer-exchange",
     "packages/dns-discovery",
     "packages/message-encryption",

packages/core/package.json

@@ -76,6 +76,7 @@
     "@noble/hashes": "^1.3.2",
     "@waku/enr": "^0.0.20",
     "@waku/interfaces": "0.0.21",
+    "@waku/message-hash": "^0.1.10",
     "@waku/proto": "0.0.6",
     "@waku/utils": "0.0.14",
     "debug": "^4.3.4",

packages/core/src/lib/base_protocol.ts

@@ -17,13 +17,16 @@ import {
 import { filterPeersByDiscovery } from "./filterPeers.js";
 import { StreamManager } from "./stream_manager.js";
 
+const DEFAULT_NUM_PEERS_TO_USE = 3;
+
 /**
  * A class with predefined helpers, to be used as a base to implement Waku
  * Protocols.
  */
 export class BaseProtocol implements IBaseProtocol {
   public readonly addLibp2pEventListener: Libp2p["addEventListener"];
   public readonly removeLibp2pEventListener: Libp2p["removeEventListener"];
+  readonly numPeersToUse: number;
   protected streamManager: StreamManager;
   protected pubsubTopics: PubsubTopic[];
 
@@ -35,6 +38,8 @@ export class BaseProtocol implements IBaseProtocol {
   ) {
     this.pubsubTopics = this.initializePubsubTopic(options);
 
+    this.numPeersToUse = options?.numPeersToUse ?? DEFAULT_NUM_PEERS_TO_USE;
+
     this.addLibp2pEventListener = components.events.addEventListener.bind(
       components.events
     );
@@ -124,6 +129,12 @@ export class BaseProtocol implements IBaseProtocol {
       );
     }
 
+    if (sortedFilteredPeers.length < numPeers) {
+      this.log.warn(
+        `Only ${sortedFilteredPeers.length} peers found. Requested ${numPeers}.`
+      );
+    }
+
     return sortedFilteredPeers;
   }