From 477c2a5918f2f75cd2c14bc6ed75e1687c5a09b4 Mon Sep 17 00:00:00 2001 From: Danish Arora <35004822+danisharora099@users.noreply.github.com> Date: Fri, 19 Jan 2024 20:42:52 +0530 Subject: [PATCH] feat!: protocols filter peers as per configured shard (#1756) * merge: master * fix: tests * update: interfafces * rm: comments * metadata: store peerIdStr instead of peerId * chore(utils): move fast-utils to dev deps * fix: allow autosharding nodes to get peers (#1785) * fix: merge * fix: build * fix: failing tests from master merge --------- Co-authored-by: Arseniy Klempner --- package-lock.json | 2 + packages/core/src/index.ts | 3 +- packages/core/src/lib/base_protocol.ts | 41 +- packages/core/src/lib/connection_manager.ts | 5 +- packages/core/src/lib/filter/index.ts | 7 +- packages/core/src/lib/filterPeers.spec.ts | 8 +- packages/core/src/lib/filterPeers.ts | 6 +- packages/core/src/lib/light_push/index.ts | 6 +- packages/core/src/lib/metadata/index.ts | 51 ++- packages/core/src/lib/store/index.ts | 7 +- packages/core/src/lib/wait_for_remote_peer.ts | 67 ++- packages/core/src/lib/waku.ts | 6 +- packages/dns-discovery/src/dns_discovery.ts | 3 +- packages/enr/src/enr.spec.ts | 1 - packages/enr/src/index.ts | 1 - packages/enr/src/raw_enr.ts | 2 +- packages/interfaces/src/metadata.ts | 7 +- packages/interfaces/src/protocols.ts | 4 +- packages/interfaces/src/waku.ts | 4 +- .../src/waku_peer_exchange_discovery.ts | 3 +- packages/tests/tests/filter/utils.ts | 3 +- packages/tests/tests/getPeers.spec.ts | 387 ++++++++++++++++++ .../single_node/multiple_pubsub.node.spec.ts | 39 +- packages/tests/tests/light-push/utils.ts | 7 +- packages/tests/tests/metadata.spec.ts | 2 +- .../tests/tests/store/multiple_pubsub.spec.ts | 32 +- packages/tests/tests/store/utils.ts | 18 +- .../tests/wait_for_remote_peer.node.spec.ts | 20 +- packages/utils/package.json | 1 + packages/utils/src/common/index.ts | 1 + .../src/common}/relay_shard_codec.spec.ts | 0 .../src/common}/relay_shard_codec.ts | 0 packages/utils/src/common/sharding.ts | 22 + packages/utils/src/libp2p/index.ts | 27 +- 34 files changed, 664 insertions(+), 129 deletions(-) rename packages/{enr/src => utils/src/common}/relay_shard_codec.spec.ts (100%) rename packages/{enr/src => utils/src/common}/relay_shard_codec.ts (100%) diff --git a/package-lock.json b/package-lock.json index a32f89de4c..c853cea8a0 100644 --- a/package-lock.json +++ b/package-lock.json @@ -28010,6 +28010,7 @@ "@rollup/plugin-node-resolve": "^15.2.3", "@waku/build-utils": "*", "cspell": "^7.3.2", + "fast-check": "^3.14.0", "npm-run-all": "^4.1.5", "rollup": "^4.9.5" }, @@ -32135,6 +32136,7 @@ "chai": "^4.3.10", "cspell": "^7.3.2", "debug": "^4.3.4", + "fast-check": "^3.14.0", "npm-run-all": "^4.1.5", "rollup": "^4.9.5", "uint8arrays": "^4.0.4" diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index c7f7462ca4..d20bdc977f 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -14,8 +14,7 @@ export * as waku_filter from "./lib/filter/index.js"; export { wakuFilter, FilterCodecs } from "./lib/filter/index.js"; export * as waku_light_push from "./lib/light_push/index.js"; -export { LightPushCodec } from "./lib/light_push/index.js"; -export { wakuLightPush } from "./lib/light_push/index.js"; +export { LightPushCodec, wakuLightPush } from "./lib/light_push/index.js"; export * as waku_store from "./lib/store/index.js"; diff --git a/packages/core/src/lib/base_protocol.ts b/packages/core/src/lib/base_protocol.ts index 5fb9d659ed..6abe59b561 100644 --- a/packages/core/src/lib/base_protocol.ts +++ b/packages/core/src/lib/base_protocol.ts @@ -9,7 +9,7 @@ import type { import { DefaultPubsubTopic } from "@waku/interfaces"; import { Logger, shardInfoToPubsubTopics } from "@waku/utils"; import { - getConnectedPeersForProtocol, + getConnectedPeersForProtocolAndShard, getPeersForProtocol, sortPeersByLatency } from "@waku/utils/libp2p"; @@ -25,12 +25,16 @@ export class BaseProtocol implements IBaseProtocol { public readonly addLibp2pEventListener: Libp2p["addEventListener"]; public readonly removeLibp2pEventListener: Libp2p["removeEventListener"]; protected streamManager: StreamManager; + protected pubsubTopics: PubsubTopic[]; constructor( public multicodec: string, private components: Libp2pComponents, - private log: Logger + private log: Logger, + private options?: ProtocolCreateOptions ) { + this.pubsubTopics = this.initializePubsubTopic(options); + this.addLibp2pEventListener = components.events.addEventListener.bind( components.events ); @@ -59,10 +63,19 @@ export class BaseProtocol implements IBaseProtocol { * the class protocol. Waku may or may not be currently connected to these * peers. */ - public async peers(): Promise { + public async allPeers(): Promise { return getPeersForProtocol(this.peerStore, [this.multicodec]); } + public async connectedPeers(): Promise { + const peers = await this.allPeers(); + return peers.filter((peer) => { + return ( + this.components.connectionManager.getConnections(peer.id).length > 0 + ); + }); + } + /** * Retrieves a list of connected peers that support the protocol. The list is sorted by latency. * @@ -83,16 +96,18 @@ export class BaseProtocol implements IBaseProtocol { numPeers: 0 } ): Promise { - // Retrieve all connected peers that support the protocol - const allPeersForProtocol = await getConnectedPeersForProtocol( - this.components.connectionManager.getConnections(), - this.peerStore, - [this.multicodec] - ); + // Retrieve all connected peers that support the protocol & shard (if configured) + const connectedPeersForProtocolAndShard = + await getConnectedPeersForProtocolAndShard( + this.components.connectionManager.getConnections(), + this.peerStore, + [this.multicodec], + this.options?.shardInfo + ); // Filter the peers based on discovery & number of peers requested - const filteredPeers = await filterPeersByDiscovery( - allPeersForProtocol, + const filteredPeers = filterPeersByDiscovery( + connectedPeersForProtocolAndShard, numPeers, maxBootstrapPeers ); @@ -112,7 +127,9 @@ export class BaseProtocol implements IBaseProtocol { return sortedFilteredPeers; } - initializePubsubTopic(options?: ProtocolCreateOptions): PubsubTopic[] { + private initializePubsubTopic( + options?: ProtocolCreateOptions + ): PubsubTopic[] { return ( options?.pubsubTopics ?? (options?.shardInfo diff --git a/packages/core/src/lib/connection_manager.ts b/packages/core/src/lib/connection_manager.ts index 242e5318a7..8a0ef17a26 100644 --- a/packages/core/src/lib/connection_manager.ts +++ b/packages/core/src/lib/connection_manager.ts @@ -1,6 +1,5 @@ import type { Peer, PeerId, PeerInfo, PeerStore } from "@libp2p/interface"; import { CustomEvent, TypedEventEmitter } from "@libp2p/interface"; -import { decodeRelayShard } from "@waku/enr"; import { ConnectionManagerOptions, EConnectionStateEvents, @@ -15,7 +14,7 @@ import { ShardInfo } from "@waku/interfaces"; import { Libp2p, Tags } from "@waku/interfaces"; -import { shardInfoToPubsubTopics } from "@waku/utils"; +import { decodeRelayShard, shardInfoToPubsubTopics } from "@waku/utils"; import { Logger } from "@waku/utils"; import { KeepAliveManager } from "./keep_alive_manager.js"; @@ -377,6 +376,8 @@ export class ConnectionManager }, "peer:connect": (evt: CustomEvent): void => { void (async () => { + log.info(`Connected to peer ${evt.detail.toString()}`); + const peerId = evt.detail; this.keepAliveManager.start( diff --git a/packages/core/src/lib/filter/index.ts b/packages/core/src/lib/filter/index.ts index 5d89322321..cf8fb126dd 100644 --- a/packages/core/src/lib/filter/index.ts +++ b/packages/core/src/lib/filter/index.ts @@ -258,7 +258,6 @@ class Subscription { } class Filter extends BaseProtocol implements IReceiver { - private readonly pubsubTopics: PubsubTopic[] = []; private activeSubscriptions = new Map(); private readonly NUM_PEERS_PROTOCOL = 1; @@ -279,9 +278,7 @@ class Filter extends BaseProtocol implements IReceiver { } constructor(libp2p: Libp2p, options?: ProtocolCreateOptions) { - super(FilterCodecs.SUBSCRIBE, libp2p.components, log); - - this.pubsubTopics = this.initializePubsubTopic(options); + super(FilterCodecs.SUBSCRIBE, libp2p.components, log, options); libp2p.handle(FilterCodecs.PUSH, this.onRequest.bind(this)).catch((e) => { log.error("Failed to register ", FilterCodecs.PUSH, e); @@ -300,8 +297,6 @@ class Filter extends BaseProtocol implements IReceiver { ensurePubsubTopicIsConfigured(pubsubTopic, this.pubsubTopics); - //TODO: get a relevant peer for the topic/shard - // https://github.com/waku-org/js-waku/pull/1586#discussion_r1336428230 const peer = ( await this.getPeers({ maxBootstrapPeers: 1, diff --git a/packages/core/src/lib/filterPeers.spec.ts b/packages/core/src/lib/filterPeers.spec.ts index 12e2449518..8ae77c3bd2 100644 --- a/packages/core/src/lib/filterPeers.spec.ts +++ b/packages/core/src/lib/filterPeers.spec.ts @@ -27,7 +27,7 @@ describe("filterPeersByDiscovery function", function () { } ] as unknown as Peer[]; - const result = await filterPeersByDiscovery(mockPeers, 0, 10); + const result = filterPeersByDiscovery(mockPeers, 0, 10); expect(result.length).to.deep.equal(mockPeers.length); }); @@ -56,7 +56,7 @@ describe("filterPeersByDiscovery function", function () { } ] as unknown as Peer[]; - const result = await filterPeersByDiscovery(mockPeers, 0, 0); + const result = filterPeersByDiscovery(mockPeers, 0, 0); // result should have no bootstrap peers, and a total of 2 peers expect(result.length).to.equal(2); @@ -95,7 +95,7 @@ describe("filterPeersByDiscovery function", function () { } ] as unknown as Peer[]; - const result = await filterPeersByDiscovery(mockPeers, 0, 1); + const result = filterPeersByDiscovery(mockPeers, 0, 1); // result should have 1 bootstrap peers, and a total of 4 peers expect(result.length).to.equal(4); @@ -134,7 +134,7 @@ describe("filterPeersByDiscovery function", function () { } ] as unknown as Peer[]; - const result = await filterPeersByDiscovery(mockPeers, 5, 2); + const result = filterPeersByDiscovery(mockPeers, 5, 2); // check that result has at least 2 bootstrap peers and no more than 5 peers expect(result.length).to.be.at.least(2); diff --git a/packages/core/src/lib/filterPeers.ts b/packages/core/src/lib/filterPeers.ts index aa0f15e897..816c3bd5b5 100644 --- a/packages/core/src/lib/filterPeers.ts +++ b/packages/core/src/lib/filterPeers.ts @@ -10,13 +10,13 @@ import { Tags } from "@waku/interfaces"; * @param peers - The list of peers to filter from. * @param numPeers - The total number of peers to retrieve. If 0, all peers are returned, irrespective of `maxBootstrapPeers`. * @param maxBootstrapPeers - The maximum number of bootstrap peers to retrieve. - * @returns A Promise that resolves to an array of peers based on the specified criteria. + * @returns An array of peers based on the specified criteria. */ -export async function filterPeersByDiscovery( +export function filterPeersByDiscovery( peers: Peer[], numPeers: number, maxBootstrapPeers: number -): Promise { +): Peer[] { // Collect the bootstrap peers up to the specified maximum let bootstrapPeers = peers .filter((peer) => peer.tags.has(Tags.BOOTSTRAP)) diff --git a/packages/core/src/lib/light_push/index.ts b/packages/core/src/lib/light_push/index.ts index b918a2105e..1ba60fdafb 100644 --- a/packages/core/src/lib/light_push/index.ts +++ b/packages/core/src/lib/light_push/index.ts @@ -5,7 +5,6 @@ import { IMessage, Libp2p, ProtocolCreateOptions, - PubsubTopic, SendError, SendResult } from "@waku/interfaces"; @@ -43,12 +42,10 @@ type PreparePushMessageResult = * Implements the [Waku v2 Light Push protocol](https://rfc.vac.dev/spec/19/). */ class LightPush extends BaseProtocol implements ILightPush { - private readonly pubsubTopics: PubsubTopic[]; private readonly NUM_PEERS_PROTOCOL = 1; constructor(libp2p: Libp2p, options?: ProtocolCreateOptions) { - super(LightPushCodec, libp2p.components, log); - this.pubsubTopics = this.initializePubsubTopic(options); + super(LightPushCodec, libp2p.components, log, options); } private async preparePushMessage( @@ -107,7 +104,6 @@ class LightPush extends BaseProtocol implements ILightPush { }; } - //TODO: get a relevant peer for the topic/shard const peers = await this.getPeers({ maxBootstrapPeers: 1, numPeers: this.NUM_PEERS_PROTOCOL diff --git a/packages/core/src/lib/metadata/index.ts b/packages/core/src/lib/metadata/index.ts index 6435c5167f..59197357a2 100644 --- a/packages/core/src/lib/metadata/index.ts +++ b/packages/core/src/lib/metadata/index.ts @@ -1,14 +1,14 @@ import type { PeerId } from "@libp2p/interface"; import { IncomingStreamData } from "@libp2p/interface"; -import { encodeRelayShard } from "@waku/enr"; import type { IMetadata, Libp2pComponents, + PeerIdStr, ShardInfo, ShardingParams } from "@waku/interfaces"; import { proto_metadata } from "@waku/proto"; -import { Logger } from "@waku/utils"; +import { encodeRelayShard, Logger } from "@waku/utils"; import all from "it-all"; import * as lp from "it-length-prefixed"; import { pipe } from "it-pipe"; @@ -20,13 +20,16 @@ const log = new Logger("metadata"); export const MetadataCodec = "/vac/waku/metadata/1.0.0"; -class Metadata extends BaseProtocol { - private readonly shardInfo: ShardingParams; +class Metadata extends BaseProtocol implements IMetadata { private libp2pComponents: Libp2pComponents; - constructor(shardInfo: ShardingParams, libp2p: Libp2pComponents) { - super(MetadataCodec, libp2p.components, log); + handshakesConfirmed: Set = new Set(); + + constructor( + public shardInfo: ShardingParams, + libp2p: Libp2pComponents + ) { + super(MetadataCodec, libp2p.components, log, shardInfo && { shardInfo }); this.libp2pComponents = libp2p; - this.shardInfo = shardInfo; void libp2p.registrar.handle(MetadataCodec, (streamData) => { void this.onRequest(streamData); }); @@ -53,12 +56,10 @@ class Metadata extends BaseProtocol { const remoteShardInfoResponse = this.decodeMetadataResponse(encodedResponse); - // add or update the shardInfo to peer store - await this.libp2pComponents.peerStore.merge(connection.remotePeer, { - metadata: { - shardInfo: encodeRelayShard(remoteShardInfoResponse) - } - }); + await this.savePeerShardInfo( + connection.remotePeer, + remoteShardInfoResponse + ); } catch (error) { log.error("Error handling metadata request", error); } @@ -87,9 +88,19 @@ class Metadata extends BaseProtocol { const decodedResponse = this.decodeMetadataResponse(encodedResponse); + await this.savePeerShardInfo(peerId, decodedResponse); + return decodedResponse; } + public async confirmOrAttemptHandshake(peerId: PeerId): Promise { + if (this.handshakesConfirmed.has(peerId.toString())) return; + + await this.query(peerId); + + return; + } + private decodeMetadataResponse(encodedResponse: Uint8ArrayList[]): ShardInfo { const bytes = new Uint8ArrayList(); @@ -104,6 +115,20 @@ class Metadata extends BaseProtocol { return response; } + + private async savePeerShardInfo( + peerId: PeerId, + shardInfo: ShardInfo + ): Promise { + // add or update the shardInfo to peer store + await this.libp2pComponents.peerStore.merge(peerId, { + metadata: { + shardInfo: encodeRelayShard(shardInfo) + } + }); + + this.handshakesConfirmed.add(peerId.toString()); + } } export function wakuMetadata( diff --git a/packages/core/src/lib/store/index.ts b/packages/core/src/lib/store/index.ts index 840c689620..d5ca1c1b47 100644 --- a/packages/core/src/lib/store/index.ts +++ b/packages/core/src/lib/store/index.ts @@ -6,8 +6,7 @@ import { IDecoder, IStore, Libp2p, - ProtocolCreateOptions, - PubsubTopic + ProtocolCreateOptions } from "@waku/interfaces"; import { proto_store as proto } from "@waku/proto"; import { ensurePubsubTopicIsConfigured, isDefined } from "@waku/utils"; @@ -74,12 +73,10 @@ export interface QueryOptions { * The Waku Store protocol can be used to retrieved historical messages. */ class Store extends BaseProtocol implements IStore { - private readonly pubsubTopics: PubsubTopic[]; private readonly NUM_PEERS_PROTOCOL = 1; constructor(libp2p: Libp2p, options?: ProtocolCreateOptions) { - super(StoreCodec, libp2p.components, log); - this.pubsubTopics = this.initializePubsubTopic(options); + super(StoreCodec, libp2p.components, log, options); } /** diff --git a/packages/core/src/lib/wait_for_remote_peer.ts b/packages/core/src/lib/wait_for_remote_peer.ts index 68e025cb83..183e81d75c 100644 --- a/packages/core/src/lib/wait_for_remote_peer.ts +++ b/packages/core/src/lib/wait_for_remote_peer.ts @@ -1,9 +1,8 @@ import type { IdentifyResult } from "@libp2p/interface"; -import type { IBaseProtocol, IRelay, Waku } from "@waku/interfaces"; +import type { IBaseProtocol, IMetadata, IRelay, Waku } from "@waku/interfaces"; import { Protocols } from "@waku/interfaces"; import { Logger } from "@waku/utils"; import { pEvent } from "p-event"; - const log = new Logger("wait-for-remote-peer"); /** @@ -32,6 +31,11 @@ export async function waitForRemotePeer( ): Promise { protocols = protocols ?? getEnabledProtocols(waku); + const isShardingEnabled = waku.shardInfo !== undefined; + const metadataService = isShardingEnabled + ? waku.libp2p.services.metadata + : undefined; + if (!waku.isStarted()) return Promise.reject("Waku node is not started"); const promises = []; @@ -45,19 +49,19 @@ export async function waitForRemotePeer( if (protocols.includes(Protocols.Store)) { if (!waku.store) throw new Error("Cannot wait for Store peer: protocol not mounted"); - promises.push(waitForConnectedPeer(waku.store)); + promises.push(waitForConnectedPeer(waku.store, metadataService)); } if (protocols.includes(Protocols.LightPush)) { if (!waku.lightPush) throw new Error("Cannot wait for LightPush peer: protocol not mounted"); - promises.push(waitForConnectedPeer(waku.lightPush)); + promises.push(waitForConnectedPeer(waku.lightPush, metadataService)); } if (protocols.includes(Protocols.Filter)) { if (!waku.filter) throw new Error("Cannot wait for Filter peer: protocol not mounted"); - promises.push(waitForConnectedPeer(waku.filter)); + promises.push(waitForConnectedPeer(waku.filter, metadataService)); } if (timeoutMs) { @@ -73,21 +77,62 @@ export async function waitForRemotePeer( /** * Wait for a peer with the given protocol to be connected. + * If sharding is enabled on the node, it will also wait for the peer to be confirmed by the metadata service. */ -async function waitForConnectedPeer(protocol: IBaseProtocol): Promise { +async function waitForConnectedPeer( + protocol: IBaseProtocol, + metadataService?: IMetadata +): Promise { const codec = protocol.multicodec; - const peers = await protocol.peers(); + const peers = await protocol.connectedPeers(); if (peers.length) { - log.info(`${codec} peer found: `, peers[0].id.toString()); - return; + if (!metadataService) { + log.info(`${codec} peer found: `, peers[0].id.toString()); + return; + } + + // once a peer is connected, we need to confirm the metadata handshake with at least one of those peers if sharding is enabled + try { + await Promise.any( + peers.map((peer) => metadataService.confirmOrAttemptHandshake(peer.id)) + ); + return; + } catch (e) { + if ((e as any).code === "ERR_CONNECTION_BEING_CLOSED") + log.error( + `Connection with the peer was closed and possibly because it's on a different shard. Error: ${e}` + ); + + log.error(`Error waiting for handshake confirmation: ${e}`); + } } + log.info(`Waiting for ${codec} peer`); + + // else we'll just wait for the next peer to connect await new Promise((resolve) => { const cb = (evt: CustomEvent): void => { if (evt.detail?.protocols?.includes(codec)) { - protocol.removeLibp2pEventListener("peer:identify", cb); - resolve(); + if (metadataService) { + metadataService + .confirmOrAttemptHandshake(evt.detail.peerId) + .then(() => { + protocol.removeLibp2pEventListener("peer:identify", cb); + resolve(); + }) + .catch((e) => { + if (e.code === "ERR_CONNECTION_BEING_CLOSED") + log.error( + `Connection with the peer was closed and possibly because it's on a different shard. Error: ${e}` + ); + + log.error(`Error waiting for handshake confirmation: ${e}`); + }); + } else { + protocol.removeLibp2pEventListener("peer:identify", cb); + resolve(); + } } }; protocol.addLibp2pEventListener("peer:identify", cb); diff --git a/packages/core/src/lib/waku.ts b/packages/core/src/lib/waku.ts index 5fd3dfc9b3..0671386e2a 100644 --- a/packages/core/src/lib/waku.ts +++ b/packages/core/src/lib/waku.ts @@ -57,7 +57,7 @@ export class WakuNode implements Waku { options: WakuOptions, pubsubTopics: PubsubTopic[] = [], libp2p: Libp2p, - pubsubShardInfo?: ShardingParams, + private pubsubShardInfo?: ShardingParams, store?: (libp2p: Libp2p) => IStore, lightPush?: (libp2p: Libp2p) => ILightPush, filter?: (libp2p: Libp2p) => IFilter, @@ -110,6 +110,10 @@ export class WakuNode implements Waku { ); } + get shardInfo(): ShardingParams | undefined { + return this.pubsubShardInfo; + } + /** * Dials to the provided peer. * diff --git a/packages/dns-discovery/src/dns_discovery.ts b/packages/dns-discovery/src/dns_discovery.ts index 0e758068a4..b58960a7d6 100644 --- a/packages/dns-discovery/src/dns_discovery.ts +++ b/packages/dns-discovery/src/dns_discovery.ts @@ -6,14 +6,13 @@ import { } from "@libp2p/interface"; import { peerDiscoverySymbol as symbol } from "@libp2p/interface"; import type { PeerInfo } from "@libp2p/interface"; -import { encodeRelayShard } from "@waku/enr"; import type { DnsDiscOptions, DnsDiscoveryComponents, IEnr, NodeCapabilityCount } from "@waku/interfaces"; -import { Logger } from "@waku/utils"; +import { encodeRelayShard, Logger } from "@waku/utils"; import { DEFAULT_BOOTSTRAP_TAG_NAME, diff --git a/packages/enr/src/enr.spec.ts b/packages/enr/src/enr.spec.ts index f04a39e239..1665e4015a 100644 --- a/packages/enr/src/enr.spec.ts +++ b/packages/enr/src/enr.spec.ts @@ -393,7 +393,6 @@ describe("ENR", function () { it("should properly create peer info with all multiaddrs", () => { const peerInfo = enr.peerInfo!; - console.log(peerInfo); expect(peerInfo.id.toString()).to.equal(peerId.toString()); expect(peerInfo.multiaddrs.length).to.equal(5); expect(peerInfo.multiaddrs.map((ma) => ma.toString())).to.contain( diff --git a/packages/enr/src/index.ts b/packages/enr/src/index.ts index 4835e900ba..d8b6fb5481 100644 --- a/packages/enr/src/index.ts +++ b/packages/enr/src/index.ts @@ -5,4 +5,3 @@ export * from "./enr.js"; export * from "./peer_id.js"; export * from "./waku2_codec.js"; export * from "./crypto.js"; -export * from "./relay_shard_codec.js"; diff --git a/packages/enr/src/raw_enr.ts b/packages/enr/src/raw_enr.ts index 252d63a4a4..b7c607880e 100644 --- a/packages/enr/src/raw_enr.ts +++ b/packages/enr/src/raw_enr.ts @@ -10,11 +10,11 @@ import type { ShardInfo, Waku2 } from "@waku/interfaces"; +import { decodeRelayShard } from "@waku/utils"; import { bytesToUtf8 } from "@waku/utils/bytes"; import { ERR_INVALID_ID } from "./constants.js"; import { decodeMultiaddrs, encodeMultiaddrs } from "./multiaddrs_codec.js"; -import { decodeRelayShard } from "./relay_shard_codec.js"; import { decodeWaku2, encodeWaku2 } from "./waku2_codec.js"; export class RawEnr extends Map { diff --git a/packages/interfaces/src/metadata.ts b/packages/interfaces/src/metadata.ts index b2c9e77e26..120e755fb6 100644 --- a/packages/interfaces/src/metadata.ts +++ b/packages/interfaces/src/metadata.ts @@ -1,8 +1,11 @@ import type { PeerId } from "@libp2p/interface"; import type { ShardInfo } from "./enr.js"; -import type { IBaseProtocol } from "./protocols.js"; +import type { IBaseProtocol, ShardingParams } from "./protocols.js"; -export interface IMetadata extends IBaseProtocol { +// IMetadata always has shardInfo defined while it is optionally undefined in IBaseProtocol +export interface IMetadata extends Omit { + shardInfo: ShardingParams; + confirmOrAttemptHandshake(peerId: PeerId): Promise; query(peerId: PeerId): Promise; } diff --git a/packages/interfaces/src/protocols.ts b/packages/interfaces/src/protocols.ts index b5a29be95e..712ec9fe8e 100644 --- a/packages/interfaces/src/protocols.ts +++ b/packages/interfaces/src/protocols.ts @@ -15,9 +15,11 @@ export enum Protocols { } export interface IBaseProtocol { + shardInfo?: ShardInfo; multicodec: string; peerStore: PeerStore; - peers: () => Promise; + allPeers: () => Promise; + connectedPeers: () => Promise; addLibp2pEventListener: Libp2p["addEventListener"]; removeLibp2pEventListener: Libp2p["removeEventListener"]; } diff --git a/packages/interfaces/src/waku.ts b/packages/interfaces/src/waku.ts index ffb6118ca3..260c806235 100644 --- a/packages/interfaces/src/waku.ts +++ b/packages/interfaces/src/waku.ts @@ -5,7 +5,7 @@ import { IConnectionManager } from "./connection_manager.js"; import type { IFilter } from "./filter.js"; import type { Libp2p } from "./libp2p.js"; import type { ILightPush } from "./light_push.js"; -import { Protocols } from "./protocols.js"; +import { Protocols, ShardingParams } from "./protocols.js"; import type { IRelay } from "./relay.js"; import type { IStore } from "./store.js"; @@ -16,6 +16,8 @@ export interface Waku { filter?: IFilter; lightPush?: ILightPush; + shardInfo?: ShardingParams; + connectionManager: IConnectionManager; dial(peer: PeerId | Multiaddr, protocols?: Protocols[]): Promise; diff --git a/packages/peer-exchange/src/waku_peer_exchange_discovery.ts b/packages/peer-exchange/src/waku_peer_exchange_discovery.ts index 2a8409433e..5690c13964 100644 --- a/packages/peer-exchange/src/waku_peer_exchange_discovery.ts +++ b/packages/peer-exchange/src/waku_peer_exchange_discovery.ts @@ -7,9 +7,8 @@ import type { PeerId, PeerInfo } from "@libp2p/interface"; -import { encodeRelayShard } from "@waku/enr"; import { Libp2pComponents, Tags } from "@waku/interfaces"; -import { Logger } from "@waku/utils"; +import { encodeRelayShard, Logger } from "@waku/utils"; import { PeerExchangeCodec, WakuPeerExchange } from "./waku_peer_exchange.js"; diff --git a/packages/tests/tests/filter/utils.ts b/packages/tests/tests/filter/utils.ts index 06ed3950d3..bcc3b26d6f 100644 --- a/packages/tests/tests/filter/utils.ts +++ b/packages/tests/tests/filter/utils.ts @@ -55,7 +55,8 @@ export async function runNodes( filter: true, lightpush: true, relay: true, - pubsubTopic: pubsubTopics + pubsubTopic: pubsubTopics, + ...(shardInfo && { clusterId: shardInfo.clusterId }) }, { retries: 3 } ); diff --git a/packages/tests/tests/getPeers.spec.ts b/packages/tests/tests/getPeers.spec.ts index 35e96192c1..6da9d27b44 100644 --- a/packages/tests/tests/getPeers.spec.ts +++ b/packages/tests/tests/getPeers.spec.ts @@ -1,16 +1,403 @@ import type { Connection, Peer, PeerStore } from "@libp2p/interface"; import { createSecp256k1PeerId } from "@libp2p/peer-id-factory"; +import { LightPushCodec, waitForRemotePeer } from "@waku/core"; import { + ContentTopicInfo, createLightNode, Libp2pComponents, type LightNode, + Protocols, + ShardInfo, Tags, utf8ToBytes } from "@waku/sdk"; +import { shardInfoToPubsubTopics } from "@waku/utils"; +import { getConnectedPeersForProtocolAndShard } from "@waku/utils/libp2p"; import { expect } from "chai"; import fc from "fast-check"; import Sinon from "sinon"; +import { makeLogFileName, ServiceNode, tearDownNodes } from "../src/index.js"; + +describe("getConnectedPeersForProtocolAndShard", function () { + let waku: LightNode; + let serviceNode1: ServiceNode; + let serviceNode2: ServiceNode; + const contentTopic = "/test/2/waku-light-push/utf8"; + + this.beforeEach(async function () { + this.timeout(15000); + serviceNode1 = new ServiceNode(makeLogFileName(this) + "1"); + serviceNode2 = new ServiceNode(makeLogFileName(this) + "2"); + }); + + afterEach(async function () { + this.timeout(15000); + await tearDownNodes([serviceNode1, serviceNode2], waku); + }); + + it("same cluster, same shard: nodes connect", async function () { + this.timeout(15000); + + const shardInfo: ShardInfo = { + clusterId: 1, + shards: [1] + }; + + await serviceNode1.start({ + discv5Discovery: true, + peerExchange: true, + clusterId: shardInfo.clusterId, + pubsubTopic: shardInfoToPubsubTopics(shardInfo), + lightpush: true, + relay: true + }); + + const serviceNodeMa = await serviceNode1.getMultiaddrWithId(); + + waku = await createLightNode({ shardInfo }); + await waku.start(); + await waku.libp2p.dialProtocol(serviceNodeMa, LightPushCodec); + await waitForRemotePeer(waku, [Protocols.LightPush]); + const peers = await getConnectedPeersForProtocolAndShard( + waku.libp2p.getConnections(), + waku.libp2p.peerStore, + waku.libp2p.getProtocols(), + shardInfo + ); + expect(peers.length).to.be.greaterThan(0); + }); + + it("same cluster, different shard: nodes connect", async function () { + this.timeout(15000); + + const shardInfo: ShardInfo = { + clusterId: 1, + shards: [1] + }; + + const shardInfoServiceNode: ShardInfo = { + clusterId: 1, + shards: [2] + }; + + await serviceNode1.start({ + discv5Discovery: true, + peerExchange: true, + clusterId: shardInfoServiceNode.clusterId, + pubsubTopic: shardInfoToPubsubTopics(shardInfoServiceNode), + lightpush: true, + relay: true + }); + + const serviceNodeMa = await serviceNode1.getMultiaddrWithId(); + + waku = await createLightNode({ shardInfo }); + await waku.libp2p.dialProtocol(serviceNodeMa, LightPushCodec); + await waku.start(); + await waitForRemotePeer(waku, [Protocols.LightPush]); + + const peers = await getConnectedPeersForProtocolAndShard( + waku.libp2p.getConnections(), + waku.libp2p.peerStore, + waku.libp2p.getProtocols(), + shardInfo + ); + expect(peers.length).to.be.greaterThan(0); + }); + + it("different cluster, same shard: nodes don't connect", async function () { + this.timeout(15000); + + const shardInfo1: ShardInfo = { + clusterId: 1, + shards: [1] + }; + + const shardInfo2: ShardInfo = { + clusterId: 2, + shards: [1] + }; + + // we start one node in a separate cluster + await serviceNode1.start({ + discv5Discovery: true, + peerExchange: true, + clusterId: shardInfo1.clusterId, + pubsubTopic: shardInfoToPubsubTopics(shardInfo1), + lightpush: true, + relay: true + }); + + // and another node in the same cluster cluster as our node + await serviceNode2.start({ + discv5Discovery: true, + peerExchange: true, + clusterId: shardInfo2.clusterId, + pubsubTopic: shardInfoToPubsubTopics(shardInfo2), + lightpush: true, + relay: true + }); + + const serviceNode1Ma = await serviceNode1.getMultiaddrWithId(); + const serviceNode2Ma = await serviceNode2.getMultiaddrWithId(); + + waku = await createLightNode({ shardInfo: shardInfo2 }); + await waku.libp2p.dialProtocol(serviceNode1Ma, LightPushCodec); + await waku.libp2p.dialProtocol(serviceNode2Ma, LightPushCodec); + + await waku.start(); + await waitForRemotePeer(waku, [Protocols.LightPush]); + + const peers = await getConnectedPeersForProtocolAndShard( + waku.libp2p.getConnections(), + waku.libp2p.peerStore, + waku.libp2p.getProtocols(), + shardInfo2 + ); + expect(peers.length).to.be.equal(1); + }); + + it("different cluster, different shard: nodes don't connect", async function () { + this.timeout(15000); + + const shardInfo1: ShardInfo = { + clusterId: 1, + shards: [1] + }; + + const shardInfo2: ShardInfo = { + clusterId: 2, + shards: [2] + }; + + // we start one node in a separate cluster + await serviceNode1.start({ + discv5Discovery: true, + peerExchange: true, + clusterId: shardInfo1.clusterId, + pubsubTopic: shardInfoToPubsubTopics(shardInfo1), + lightpush: true, + relay: true + }); + + // and another node in the same cluster cluster as our node + const serviceNode2 = new ServiceNode(makeLogFileName(this) + "2"); + await serviceNode2.start({ + discv5Discovery: true, + peerExchange: true, + clusterId: shardInfo2.clusterId, + pubsubTopic: shardInfoToPubsubTopics(shardInfo2), + lightpush: true, + relay: true + }); + + const serviceNodeMa1 = await serviceNode1.getMultiaddrWithId(); + const serviceNodeMa2 = await serviceNode2.getMultiaddrWithId(); + + waku = await createLightNode({ shardInfo: shardInfo2 }); + await waku.libp2p.dialProtocol(serviceNodeMa1, LightPushCodec); + await waku.libp2p.dialProtocol(serviceNodeMa2, LightPushCodec); + await waku.start(); + await waitForRemotePeer(waku, [Protocols.LightPush]); + + const peers = await getConnectedPeersForProtocolAndShard( + waku.libp2p.getConnections(), + waku.libp2p.peerStore, + waku.libp2p.getProtocols(), + shardInfo2 + ); + expect(peers.length).to.be.equal(1); + }); + + it("same cluster, same shard: nodes connect (autosharding)", async function () { + this.timeout(15000); + + const shardInfo: ContentTopicInfo = { + clusterId: 1, + contentTopics: [contentTopic] + }; + + await serviceNode1.start({ + discv5Discovery: true, + peerExchange: true, + clusterId: shardInfo.clusterId, + pubsubTopic: shardInfoToPubsubTopics(shardInfo), + lightpush: true, + relay: true + }); + + const serviceNodeMa = await serviceNode1.getMultiaddrWithId(); + + waku = await createLightNode({ shardInfo }); + await waku.start(); + await waku.libp2p.dialProtocol(serviceNodeMa, LightPushCodec); + await waitForRemotePeer(waku, [Protocols.LightPush]); + const peers = await getConnectedPeersForProtocolAndShard( + waku.libp2p.getConnections(), + waku.libp2p.peerStore, + waku.libp2p.getProtocols(), + shardInfo + ); + expect(peers.length).to.be.greaterThan(0); + }); + + it("same cluster, different shard: nodes connect (autosharding)", async function () { + this.timeout(15000); + + const shardInfo1: ContentTopicInfo = { + clusterId: 1, + contentTopics: [contentTopic] + }; + + const shardInfo2: ContentTopicInfo = { + clusterId: 1, + contentTopics: ["/test/5/waku-light-push/utf8"] + }; + + // Separate shard + await serviceNode1.start({ + discv5Discovery: true, + peerExchange: true, + clusterId: shardInfo1.clusterId, + pubsubTopic: shardInfoToPubsubTopics(shardInfo1), + lightpush: true, + relay: true + }); + + // Same shard + await serviceNode2.start({ + discv5Discovery: true, + peerExchange: true, + clusterId: shardInfo2.clusterId, + pubsubTopic: shardInfoToPubsubTopics(shardInfo2), + lightpush: true, + relay: true + }); + + const serviceNode1Ma = await serviceNode1.getMultiaddrWithId(); + const serviceNode2Ma = await serviceNode2.getMultiaddrWithId(); + + waku = await createLightNode({ shardInfo: shardInfo2 }); + await waku.libp2p.dialProtocol(serviceNode1Ma, LightPushCodec); + await waku.libp2p.dialProtocol(serviceNode2Ma, LightPushCodec); + + await waku.start(); + await waitForRemotePeer(waku, [Protocols.LightPush]); + + const peers = await getConnectedPeersForProtocolAndShard( + waku.libp2p.getConnections(), + waku.libp2p.peerStore, + waku.libp2p.getProtocols(), + shardInfo2 + ); + expect(peers.length).to.be.equal(1); + }); + + it("different cluster, same shard: nodes don't connect (autosharding)", async function () { + this.timeout(15000); + + const shardInfo1: ContentTopicInfo = { + clusterId: 1, + contentTopics: [contentTopic] + }; + + const shardInfo2: ContentTopicInfo = { + clusterId: 2, + contentTopics: [contentTopic] + }; + + // we start one node in a separate cluster + await serviceNode1.start({ + discv5Discovery: true, + peerExchange: true, + clusterId: shardInfo1.clusterId, + pubsubTopic: shardInfoToPubsubTopics(shardInfo1), + lightpush: true, + relay: true + }); + + // and another node in the same cluster cluster as our node + await serviceNode2.start({ + discv5Discovery: true, + peerExchange: true, + clusterId: shardInfo2.clusterId, + pubsubTopic: shardInfoToPubsubTopics(shardInfo2), + lightpush: true, + relay: true + }); + + const serviceNode1Ma = await serviceNode1.getMultiaddrWithId(); + const serviceNode2Ma = await serviceNode2.getMultiaddrWithId(); + + waku = await createLightNode({ shardInfo: shardInfo2 }); + await waku.libp2p.dialProtocol(serviceNode1Ma, LightPushCodec); + await waku.libp2p.dialProtocol(serviceNode2Ma, LightPushCodec); + + await waku.start(); + await waitForRemotePeer(waku, [Protocols.LightPush]); + + const peers = await getConnectedPeersForProtocolAndShard( + waku.libp2p.getConnections(), + waku.libp2p.peerStore, + waku.libp2p.getProtocols(), + shardInfo2 + ); + expect(peers.length).to.be.equal(1); + }); + + it("different cluster, different shard: nodes don't connect (autosharding)", async function () { + this.timeout(15000); + + const shardInfo1: ContentTopicInfo = { + clusterId: 1, + contentTopics: [contentTopic] + }; + + const shardInfo2: ContentTopicInfo = { + clusterId: 2, + contentTopics: ["/test/5/waku-light-push/utf8"] + }; + + // we start one node in a separate cluster + await serviceNode1.start({ + discv5Discovery: true, + peerExchange: true, + clusterId: shardInfo1.clusterId, + pubsubTopic: shardInfoToPubsubTopics(shardInfo1), + lightpush: true, + relay: true + }); + + // and another node in the same cluster cluster as our node + const serviceNode2 = new ServiceNode(makeLogFileName(this) + "2"); + await serviceNode2.start({ + discv5Discovery: true, + peerExchange: true, + clusterId: shardInfo2.clusterId, + pubsubTopic: shardInfoToPubsubTopics(shardInfo2), + lightpush: true, + relay: true + }); + + const serviceNodeMa1 = await serviceNode1.getMultiaddrWithId(); + const serviceNodeMa2 = await serviceNode2.getMultiaddrWithId(); + + waku = await createLightNode({ shardInfo: shardInfo2 }); + await waku.libp2p.dialProtocol(serviceNodeMa1, LightPushCodec); + await waku.libp2p.dialProtocol(serviceNodeMa2, LightPushCodec); + await waku.start(); + await waitForRemotePeer(waku, [Protocols.LightPush]); + + const peers = await getConnectedPeersForProtocolAndShard( + waku.libp2p.getConnections(), + waku.libp2p.peerStore, + waku.libp2p.getProtocols(), + shardInfo2 + ); + expect(peers.length).to.be.equal(1); + }); +}); describe("getPeers", function () { let peerStore: PeerStore; let connectionManager: Libp2pComponents["connectionManager"]; diff --git a/packages/tests/tests/light-push/single_node/multiple_pubsub.node.spec.ts b/packages/tests/tests/light-push/single_node/multiple_pubsub.node.spec.ts index d7a4c2b121..a5e7745e32 100644 --- a/packages/tests/tests/light-push/single_node/multiple_pubsub.node.spec.ts +++ b/packages/tests/tests/light-push/single_node/multiple_pubsub.node.spec.ts @@ -33,10 +33,7 @@ describe("Waku Light Push : Multiple PubsubTopics", function () { clusterId: 3, shard: 1 }); - const customPubsubTopic2 = singleShardInfoToPubsubTopic({ - clusterId: 3, - shard: 2 - }); + const shardInfo: ShardInfo = { clusterId: 3, shards: [1, 2] }; const singleShardInfo1: SingleShardInfo = { clusterId: 3, shard: 1 }; const singleShardInfo2: SingleShardInfo = { clusterId: 3, shard: 2 }; @@ -57,7 +54,10 @@ describe("Waku Light Push : Multiple PubsubTopics", function () { this.timeout(15000); [nwaku, waku] = await runNodes( this, - [customPubsubTopic1, customPubsubTopic2], + [ + singleShardInfoToPubsubTopic(singleShardInfo1), + singleShardInfoToPubsubTopic(singleShardInfo2) + ], shardInfo ); messageCollector = new MessageCollector(nwaku); @@ -107,7 +107,7 @@ describe("Waku Light Push : Multiple PubsubTopics", function () { expect( await messageCollector2.waitForMessages(1, { - pubsubTopic: customPubsubTopic2 + pubsubTopic: singleShardInfoToPubsubTopic(singleShardInfo2) }) ).to.eq(true); @@ -130,9 +130,12 @@ describe("Waku Light Push : Multiple PubsubTopics", function () { filter: true, lightpush: true, relay: true, - pubsubTopic: [customPubsubTopic2] + pubsubTopic: [singleShardInfoToPubsubTopic(singleShardInfo2)], + clusterId: singleShardInfo2.clusterId }); - await nwaku2.ensureSubscriptions([customPubsubTopic2]); + await nwaku2.ensureSubscriptions([ + singleShardInfoToPubsubTopic(singleShardInfo2) + ]); await waku.dial(await nwaku2.getMultiaddrWithId()); await waitForRemotePeer(waku, [Protocols.LightPush]); @@ -147,7 +150,7 @@ describe("Waku Light Push : Multiple PubsubTopics", function () { pubsubTopic: customPubsubTopic1 })) || !(await messageCollector2.waitForMessages(1, { - pubsubTopic: customPubsubTopic2 + pubsubTopic: singleShardInfoToPubsubTopic(singleShardInfo2) })) || pushResponse1!.recipients[0].toString() === pushResponse2!.recipients[0].toString() @@ -168,7 +171,7 @@ describe("Waku Light Push : Multiple PubsubTopics", function () { messageCollector2.verifyReceivedMessage(0, { expectedMessageText: "M2", expectedContentTopic: customContentTopic2, - expectedPubsubTopic: customPubsubTopic2 + expectedPubsubTopic: singleShardInfoToPubsubTopic(singleShardInfo2) }); }); }); @@ -286,7 +289,8 @@ describe("Waku Light Push (Autosharding): Multiple PubsubTopics", function () { filter: true, lightpush: true, relay: true, - pubsubTopic: [autoshardingPubsubTopic2] + pubsubTopic: [autoshardingPubsubTopic2], + clusterId: shardInfo.clusterId }); await nwaku2.ensureSubscriptionsAutosharding([customContentTopic2]); await waku.dial(await nwaku2.getMultiaddrWithId()); @@ -349,10 +353,6 @@ describe("Waku Light Push (named sharding): Multiple PubsubTopics", function () customContentTopic2, clusterId ); - const contentTopicInfo: ContentTopicInfo = { - clusterId, - contentTopics: [customContentTopic1, customContentTopic2] - }; const customEncoder1 = createEncoder({ contentTopic: customContentTopic1, pubsubTopicShardInfo: { @@ -368,11 +368,10 @@ describe("Waku Light Push (named sharding): Multiple PubsubTopics", function () this.beforeEach(async function () { this.timeout(15000); - [nwaku, waku] = await runNodes( - this, - [autoshardingPubsubTopic1, autoshardingPubsubTopic2], - contentTopicInfo - ); + [nwaku, waku] = await runNodes(this, [ + autoshardingPubsubTopic1, + autoshardingPubsubTopic2 + ]); messageCollector = new MessageCollector(nwaku); nimPeerId = await nwaku.getPeerId(); }); diff --git a/packages/tests/tests/light-push/utils.ts b/packages/tests/tests/light-push/utils.ts index 7fdcac0de3..c279fa713a 100644 --- a/packages/tests/tests/light-push/utils.ts +++ b/packages/tests/tests/light-push/utils.ts @@ -24,7 +24,12 @@ export async function runNodes( ): Promise<[ServiceNode, LightNode]> { const nwaku = new ServiceNode(makeLogFileName(context)); await nwaku.start( - { lightpush: true, relay: true, pubsubTopic: pubsubTopics }, + { + lightpush: true, + relay: true, + pubsubTopic: pubsubTopics, + ...(shardInfo && { clusterId: shardInfo.clusterId }) + }, { retries: 3 } ); diff --git a/packages/tests/tests/metadata.spec.ts b/packages/tests/tests/metadata.spec.ts index 0996910815..1d7a003c9f 100644 --- a/packages/tests/tests/metadata.spec.ts +++ b/packages/tests/tests/metadata.spec.ts @@ -1,7 +1,7 @@ import { MetadataCodec } from "@waku/core"; -import { decodeRelayShard } from "@waku/enr"; import type { LightNode, ShardInfo } from "@waku/interfaces"; import { createLightNode } from "@waku/sdk"; +import { decodeRelayShard } from "@waku/utils"; import { shardInfoToPubsubTopics } from "@waku/utils"; import chai, { expect } from "chai"; import chaiAsPromised from "chai-as-promised"; diff --git a/packages/tests/tests/store/multiple_pubsub.spec.ts b/packages/tests/tests/store/multiple_pubsub.spec.ts index a41597e065..70f5c1acec 100644 --- a/packages/tests/tests/store/multiple_pubsub.spec.ts +++ b/packages/tests/tests/store/multiple_pubsub.spec.ts @@ -1,7 +1,10 @@ import { createDecoder, waitForRemotePeer } from "@waku/core"; import type { ContentTopicInfo, IMessage, LightNode } from "@waku/interfaces"; import { createLightNode, Protocols } from "@waku/sdk"; -import { contentTopicToPubsubTopic } from "@waku/utils"; +import { + contentTopicToPubsubTopic, + singleShardInfosToShardInfo +} from "@waku/utils"; import { expect } from "chai"; import { @@ -18,6 +21,8 @@ import { customDecoder2, customShardedPubsubTopic1, customShardedPubsubTopic2, + customShardInfo1, + customShardInfo2, processQueriedMessages, sendMessages, sendMessagesAutosharding, @@ -39,6 +44,7 @@ describe("Waku Store, custom pubsub topic", function () { await nwaku.start({ store: true, pubsubTopic: [customShardedPubsubTopic1, customShardedPubsubTopic2], + clusterId: customShardInfo1.clusterId, relay: true }); await nwaku.ensureSubscriptions([ @@ -123,6 +129,7 @@ describe("Waku Store, custom pubsub topic", function () { await nwaku2.start({ store: true, pubsubTopic: [customShardedPubsubTopic2], + clusterId: customShardInfo2.clusterId, relay: true }); await nwaku2.ensureSubscriptions([customShardedPubsubTopic2]); @@ -210,7 +217,8 @@ describe("Waku Store (Autosharding), custom pubsub topic", function () { await nwaku.start({ store: true, pubsubTopic: [autoshardingPubsubTopic1, autoshardingPubsubTopic2], - relay: true + relay: true, + clusterId }); await nwaku.ensureSubscriptionsAutosharding([ customContentTopic1, @@ -283,7 +291,8 @@ describe("Waku Store (Autosharding), custom pubsub topic", function () { await nwaku2.start({ store: true, pubsubTopic: [autoshardingPubsubTopic2], - relay: true + relay: true, + clusterId }); await nwaku2.ensureSubscriptionsAutosharding([customContentTopic2]); @@ -339,11 +348,18 @@ describe("Waku Store (named sharding), custom pubsub topic", function () { beforeEach(async function () { this.timeout(15000); + + const shardInfo = singleShardInfosToShardInfo([ + customShardInfo1, + customShardInfo2 + ]); + nwaku = new ServiceNode(makeLogFileName(this)); await nwaku.start({ store: true, relay: true, - pubsubTopic: [customShardedPubsubTopic1, customShardedPubsubTopic2] + pubsubTopic: [customShardedPubsubTopic1, customShardedPubsubTopic2], + clusterId: shardInfo.clusterId }); await nwaku.ensureSubscriptions([ customShardedPubsubTopic1, @@ -353,10 +369,7 @@ describe("Waku Store (named sharding), custom pubsub topic", function () { waku = await startAndConnectLightNode( nwaku, [customShardedPubsubTopic1, customShardedPubsubTopic2], - { - clusterId: 3, - shards: [1, 2] - } + shardInfo ); }); @@ -434,7 +447,8 @@ describe("Waku Store (named sharding), custom pubsub topic", function () { await nwaku2.start({ store: true, pubsubTopic: [customShardedPubsubTopic2], - relay: true + relay: true, + clusterId: customShardInfo2.clusterId }); await nwaku2.ensureSubscriptions([customShardedPubsubTopic2]); diff --git a/packages/tests/tests/store/utils.ts b/packages/tests/tests/store/utils.ts index 994a39c4ca..d1b88161d0 100644 --- a/packages/tests/tests/store/utils.ts +++ b/packages/tests/tests/store/utils.ts @@ -10,7 +10,8 @@ import { LightNode, Protocols, ShardInfo, - ShardingParams + ShardingParams, + type SingleShardInfo } from "@waku/interfaces"; import { createLightNode } from "@waku/sdk"; import { Logger, singleShardInfoToPubsubTopic } from "@waku/utils"; @@ -23,14 +24,13 @@ export const log = new Logger("test:store"); export const TestContentTopic = "/test/1/waku-store/utf8"; export const TestEncoder = createEncoder({ contentTopic: TestContentTopic }); export const TestDecoder = createDecoder(TestContentTopic); -export const customShardedPubsubTopic1 = singleShardInfoToPubsubTopic({ - clusterId: 3, - shard: 1 -}); -export const customShardedPubsubTopic2 = singleShardInfoToPubsubTopic({ - clusterId: 3, - shard: 2 -}); +export const customShardInfo1: SingleShardInfo = { clusterId: 3, shard: 1 }; +export const customShardedPubsubTopic1 = + singleShardInfoToPubsubTopic(customShardInfo1); + +export const customShardInfo2: SingleShardInfo = { clusterId: 3, shard: 2 }; +export const customShardedPubsubTopic2 = + singleShardInfoToPubsubTopic(customShardInfo2); export const shardInfo1: ShardInfo = { clusterId: 3, shards: [1] }; export const customContentTopic1 = "/test/2/waku-store/utf8"; export const customContentTopic2 = "/test/3/waku-store/utf8"; diff --git a/packages/tests/tests/wait_for_remote_peer.node.spec.ts b/packages/tests/tests/wait_for_remote_peer.node.spec.ts index a984c650f6..51551ba176 100644 --- a/packages/tests/tests/wait_for_remote_peer.node.spec.ts +++ b/packages/tests/tests/wait_for_remote_peer.node.spec.ts @@ -115,7 +115,9 @@ describe("Wait for remote peer", function () { await delay(1000); await waitForRemotePeer(waku2, [Protocols.Store]); - const peers = (await waku2.store.peers()).map((peer) => peer.id.toString()); + const peers = (await waku2.store.connectedPeers()).map((peer) => + peer.id.toString() + ); const nimPeerId = multiAddrWithId.getPeerId(); expect(nimPeerId).to.not.be.undefined; @@ -142,7 +144,9 @@ describe("Wait for remote peer", function () { await waku2.dial(multiAddrWithId); await waitPromise; - const peers = (await waku2.store.peers()).map((peer) => peer.id.toString()); + const peers = (await waku2.store.connectedPeers()).map((peer) => + peer.id.toString() + ); const nimPeerId = multiAddrWithId.getPeerId(); @@ -168,7 +172,7 @@ describe("Wait for remote peer", function () { await waku2.dial(multiAddrWithId); await waitForRemotePeer(waku2, [Protocols.LightPush]); - const peers = (await waku2.lightPush.peers()).map((peer) => + const peers = (await waku2.lightPush.connectedPeers()).map((peer) => peer.id.toString() ); @@ -196,7 +200,7 @@ describe("Wait for remote peer", function () { await waku2.dial(multiAddrWithId); await waitForRemotePeer(waku2, [Protocols.Filter]); - const peers = (await waku2.filter.peers()).map((peer) => + const peers = (await waku2.filter.connectedPeers()).map((peer) => peer.id.toString() ); @@ -228,14 +232,14 @@ describe("Wait for remote peer", function () { Protocols.LightPush ]); - const filterPeers = (await waku2.filter.peers()).map((peer) => + const filterPeers = (await waku2.filter.connectedPeers()).map((peer) => peer.id.toString() ); - const storePeers = (await waku2.store.peers()).map((peer) => + const storePeers = (await waku2.store.connectedPeers()).map((peer) => peer.id.toString() ); - const lightPushPeers = (await waku2.lightPush.peers()).map((peer) => - peer.id.toString() + const lightPushPeers = (await waku2.lightPush.connectedPeers()).map( + (peer) => peer.id.toString() ); const nimPeerId = multiAddrWithId.getPeerId(); diff --git a/packages/utils/package.json b/packages/utils/package.json index 13d296ac56..1f764d6fca 100644 --- a/packages/utils/package.json +++ b/packages/utils/package.json @@ -80,6 +80,7 @@ "@waku/build-utils": "*", "cspell": "^7.3.2", "npm-run-all": "^4.1.5", + "fast-check": "^3.14.0", "rollup": "^4.9.5" }, "files": [ diff --git a/packages/utils/src/common/index.ts b/packages/utils/src/common/index.ts index f834bc22c4..5fbc976815 100644 --- a/packages/utils/src/common/index.ts +++ b/packages/utils/src/common/index.ts @@ -5,6 +5,7 @@ export * from "./to_async_iterator.js"; export * from "./is_size_valid.js"; export * from "./sharding.js"; export * from "./push_or_init_map.js"; +export * from "./relay_shard_codec.js"; export function removeItemFromArray(arr: unknown[], value: unknown): unknown[] { const index = arr.indexOf(value); diff --git a/packages/enr/src/relay_shard_codec.spec.ts b/packages/utils/src/common/relay_shard_codec.spec.ts similarity index 100% rename from packages/enr/src/relay_shard_codec.spec.ts rename to packages/utils/src/common/relay_shard_codec.spec.ts diff --git a/packages/enr/src/relay_shard_codec.ts b/packages/utils/src/common/relay_shard_codec.ts similarity index 100% rename from packages/enr/src/relay_shard_codec.ts rename to packages/utils/src/common/relay_shard_codec.ts diff --git a/packages/utils/src/common/sharding.ts b/packages/utils/src/common/sharding.ts index f6f3cd7b34..18ee894eb4 100644 --- a/packages/utils/src/common/sharding.ts +++ b/packages/utils/src/common/sharding.ts @@ -2,6 +2,7 @@ import { sha256 } from "@noble/hashes/sha256"; import { DefaultPubsubTopic, PubsubTopic, + ShardInfo, ShardingParams, SingleShardInfo } from "@waku/interfaces"; @@ -17,11 +18,32 @@ export const singleShardInfoToPubsubTopic = ( return `/waku/2/rs/${shardInfo.clusterId}/${shardInfo.shard}`; }; +export const singleShardInfosToShardInfo = ( + singleShardInfos: SingleShardInfo[] +): ShardInfo => { + if (singleShardInfos.length === 0) throw new Error("Invalid shard"); + + const clusterIds = singleShardInfos.map((shardInfo) => shardInfo.clusterId); + if (new Set(clusterIds).size !== 1) { + throw new Error("Passed shard infos have different clusterIds"); + } + + const shards = singleShardInfos + .map((shardInfo) => shardInfo.shard) + .filter((shard): shard is number => shard !== undefined); + + return { + clusterId: singleShardInfos[0].clusterId, + shards + }; +}; + export const shardInfoToPubsubTopics = ( shardInfo: ShardingParams ): PubsubTopic[] => { if (shardInfo.clusterId === undefined) throw new Error("Cluster ID must be specified"); + if ("contentTopics" in shardInfo) { // Autosharding: explicitly defined content topics return Array.from( diff --git a/packages/utils/src/libp2p/index.ts b/packages/utils/src/libp2p/index.ts index d185b35f82..d562cc49a6 100644 --- a/packages/utils/src/libp2p/index.ts +++ b/packages/utils/src/libp2p/index.ts @@ -1,6 +1,8 @@ import type { Connection, Peer, PeerStore } from "@libp2p/interface"; +import { ShardingParams } from "@waku/interfaces"; import { bytesToUtf8 } from "../bytes/index.js"; +import { decodeRelayShard } from "../common/relay_shard_codec.js"; /** * Returns a pseudo-random peer that supports the given protocol. @@ -68,10 +70,11 @@ export async function getPeersForProtocol( return peers; } -export async function getConnectedPeersForProtocol( +export async function getConnectedPeersForProtocolAndShard( connections: Connection[], peerStore: PeerStore, - protocols: string[] + protocols: string[], + shardInfo?: ShardingParams ): Promise { const openConnections = connections.filter( (connection) => connection.status === "open" @@ -79,10 +82,24 @@ export async function getConnectedPeersForProtocol( const peerPromises = openConnections.map(async (connection) => { const peer = await peerStore.get(connection.remotePeer); - const supportsProtocol = peer.protocols.some((protocol) => - protocols.includes(protocol) + const supportsProtocol = protocols.some((protocol) => + peer.protocols.includes(protocol) ); - return supportsProtocol ? peer : null; + + if (supportsProtocol) { + if (shardInfo) { + const encodedPeerShardInfo = peer.metadata.get("shardInfo"); + const peerShardInfo = + encodedPeerShardInfo && decodeRelayShard(encodedPeerShardInfo); + + if (peerShardInfo && shardInfo.clusterId === peerShardInfo.clusterId) { + return peer; + } + } else { + return peer; + } + } + return null; }); const peersWithNulls = await Promise.all(peerPromises);