logos-messaging/js-waku
1
0
Fork 0
mirror of https://github.com/logos-messaging/js-waku.git synced 2026-01-05 23:33:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: move Peer to PeerId (#2246)

Browse Source 
* feat: move Peer to PeerId

* up tests

* update tests
This commit is contained in:
Sasha 2025-02-05 11:21:22 +01:00 committed by GitHub
parent 2a7f4b6e86
commit fc93fae873
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
15 changed files with 135 additions and 143 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
packages/core/src/lib/base_protocol.ts
View File

@ -1,5 +1,5 @@
import type { Libp2p } from "@libp2p/interface"; import type { Libp2p } from "@libp2p/interface";
import type { Peer, Stream } from "@libp2p/interface"; import type { PeerId, Stream } from "@libp2p/interface";
import type { import type {
IBaseProtocolCore, IBaseProtocolCore,
Libp2pComponents, Libp2pComponents,
@ -38,7 +38,7 @@ export class BaseProtocol implements IBaseProtocolCore {
); );
} }
protected async getStream(peer: Peer): Promise<Stream> { protected async getStream(peerId: PeerId): Promise<Stream> {
return this.streamManager.getStream(peer); return this.streamManager.getStream(peerId);
} }
} }

50
packages/core/src/lib/filter/filter.ts
View File

@ -1,4 +1,4 @@
import type { Peer, Stream } from "@libp2p/interface"; import type { PeerId, Stream } from "@libp2p/interface";
import type { IncomingStreamData } from "@libp2p/interface-internal"; import type { IncomingStreamData } from "@libp2p/interface-internal";
import { import {
type ContentTopic, type ContentTopic,
@ -53,10 +53,10 @@ export class FilterCore extends BaseProtocol implements IBaseProtocolCore {
public async subscribe( public async subscribe(
pubsubTopic: PubsubTopic, pubsubTopic: PubsubTopic,
peer: Peer, peerId: PeerId,
contentTopics: ContentTopic[] contentTopics: ContentTopic[]
): Promise<CoreProtocolResult> { ): Promise<CoreProtocolResult> {
const stream = await this.getStream(peer); const stream = await this.getStream(peerId);
const request = FilterSubscribeRpc.createSubscribeRequest( const request = FilterSubscribeRpc.createSubscribeRequest(
pubsubTopic, pubsubTopic,
@ -78,7 +78,7 @@ export class FilterCore extends BaseProtocol implements IBaseProtocolCore {
success: null, success: null,
failure: { failure: {
error: ProtocolError.GENERIC_FAIL, error: ProtocolError.GENERIC_FAIL,
peerId: peer.id peerId: peerId
} }
}; };
} }
@ -93,7 +93,7 @@ export class FilterCore extends BaseProtocol implements IBaseProtocolCore {
return { return {
failure: { failure: {
error: ProtocolError.REMOTE_PEER_REJECTED, error: ProtocolError.REMOTE_PEER_REJECTED,
peerId: peer.id peerId: peerId
}, },
success: null success: null
}; };
@ -101,28 +101,28 @@ export class FilterCore extends BaseProtocol implements IBaseProtocolCore {
return { return {
failure: null, failure: null,
success: peer.id success: peerId
}; };
} }
public async unsubscribe( public async unsubscribe(
pubsubTopic: PubsubTopic, pubsubTopic: PubsubTopic,
peer: Peer, peerId: PeerId,
contentTopics: ContentTopic[] contentTopics: ContentTopic[]
): Promise<CoreProtocolResult> { ): Promise<CoreProtocolResult> {
let stream: Stream | undefined; let stream: Stream | undefined;
try { try {
stream = await this.getStream(peer); stream = await this.getStream(peerId);
} catch (error) { } catch (error) {
log.error( log.error(
`Failed to get a stream for remote peer${peer.id.toString()}`, `Failed to get a stream for remote peer${peerId.toString()}`,
error error
); );
return { return {
success: null, success: null,
failure: { failure: {
error: ProtocolError.NO_STREAM_AVAILABLE, error: ProtocolError.NO_STREAM_AVAILABLE,
peerId: peer.id peerId: peerId
} }
}; };
} }
@ -140,22 +140,22 @@ export class FilterCore extends BaseProtocol implements IBaseProtocolCore {
success: null, success: null,
failure: { failure: {
error: ProtocolError.GENERIC_FAIL, error: ProtocolError.GENERIC_FAIL,
peerId: peer.id peerId: peerId
} }
}; };
} }
return { return {
success: peer.id, success: peerId,
failure: null failure: null
}; };
} }
public async unsubscribeAll( public async unsubscribeAll(
pubsubTopic: PubsubTopic, pubsubTopic: PubsubTopic,
peer: Peer peerId: PeerId
): Promise<CoreProtocolResult> { ): Promise<CoreProtocolResult> {
const stream = await this.getStream(peer); const stream = await this.getStream(peerId);
const request = FilterSubscribeRpc.createUnsubscribeAllRequest(pubsubTopic); const request = FilterSubscribeRpc.createUnsubscribeAllRequest(pubsubTopic);
@ -171,7 +171,7 @@ export class FilterCore extends BaseProtocol implements IBaseProtocolCore {
return { return {
failure: { failure: {
error: ProtocolError.NO_RESPONSE, error: ProtocolError.NO_RESPONSE,
peerId: peer.id peerId: peerId
}, },
success: null success: null
}; };
@ -187,7 +187,7 @@ export class FilterCore extends BaseProtocol implements IBaseProtocolCore {
return { return {
failure: { failure: {
error: ProtocolError.REMOTE_PEER_REJECTED, error: ProtocolError.REMOTE_PEER_REJECTED,
peerId: peer.id peerId: peerId
}, },
success: null success: null
}; };
@ -195,24 +195,24 @@ export class FilterCore extends BaseProtocol implements IBaseProtocolCore {
return { return {
failure: null, failure: null,
success: peer.id success: peerId
}; };
} }
public async ping(peer: Peer): Promise<CoreProtocolResult> { public async ping(peerId: PeerId): Promise<CoreProtocolResult> {
let stream: Stream | undefined; let stream: Stream | undefined;
try { try {
stream = await this.getStream(peer); stream = await this.getStream(peerId);
} catch (error) { } catch (error) {
log.error( log.error(
`Failed to get a stream for remote peer${peer.id.toString()}`, `Failed to get a stream for remote peer${peerId.toString()}`,
error error
); );
return { return {
success: null, success: null,
failure: { failure: {
error: ProtocolError.NO_STREAM_AVAILABLE, error: ProtocolError.NO_STREAM_AVAILABLE,
peerId: peer.id peerId: peerId
} }
}; };
} }
@ -234,7 +234,7 @@ export class FilterCore extends BaseProtocol implements IBaseProtocolCore {
success: null, success: null,
failure: { failure: {
error: ProtocolError.GENERIC_FAIL, error: ProtocolError.GENERIC_FAIL,
peerId: peer.id peerId: peerId
} }
}; };
} }
@ -244,7 +244,7 @@ export class FilterCore extends BaseProtocol implements IBaseProtocolCore {
success: null, success: null,
failure: { failure: {
error: ProtocolError.NO_RESPONSE, error: ProtocolError.NO_RESPONSE,
peerId: peer.id peerId: peerId
} }
}; };
} }
@ -260,12 +260,12 @@ export class FilterCore extends BaseProtocol implements IBaseProtocolCore {
success: null, success: null,
failure: { failure: {
error: ProtocolError.REMOTE_PEER_REJECTED, error: ProtocolError.REMOTE_PEER_REJECTED,
peerId: peer.id peerId: peerId
} }
}; };
} }
return { return {
success: peer.id, success: peerId,
failure: null failure: null
}; };
} }

23
packages/core/src/lib/light_push/light_push.ts
View File

@ -1,4 +1,4 @@
import type { Peer, Stream } from "@libp2p/interface"; import type { PeerId, Stream } from "@libp2p/interface";
import { import {
type CoreProtocolResult, type CoreProtocolResult,
type IBaseProtocolCore, type IBaseProtocolCore,
@ -76,11 +76,10 @@ export class LightPushCore extends BaseProtocol implements IBaseProtocolCore {
} }
} }
// TODO(weboko): use peer.id as parameter instead
public async send( public async send(
encoder: IEncoder, encoder: IEncoder,
message: IMessage, message: IMessage,
peer: Peer peerId: PeerId
): Promise<CoreProtocolResult> { ): Promise<CoreProtocolResult> {
const { query, error: preparationError } = await this.preparePushMessage( const { query, error: preparationError } = await this.preparePushMessage(
encoder, encoder,
@ -92,21 +91,21 @@ export class LightPushCore extends BaseProtocol implements IBaseProtocolCore {
success: null, success: null,
failure: { failure: {
error: preparationError, error: preparationError,
peerId: peer.id peerId
} }
}; };
} }
let stream: Stream; let stream: Stream;
try { try {
stream = await this.getStream(peer); stream = await this.getStream(peerId);
} catch (error) { } catch (error) {
log.error("Failed to get stream", error); log.error("Failed to get stream", error);
return { return {
success: null, success: null,
failure: { failure: {
error: ProtocolError.NO_STREAM_AVAILABLE, error: ProtocolError.NO_STREAM_AVAILABLE,
peerId: peer.id peerId: peerId
} }
}; };
} }
@ -126,7 +125,7 @@ export class LightPushCore extends BaseProtocol implements IBaseProtocolCore {
success: null, success: null,
failure: { failure: {
error: ProtocolError.GENERIC_FAIL, error: ProtocolError.GENERIC_FAIL,
peerId: peer.id peerId: peerId
} }
}; };
} }
@ -145,7 +144,7 @@ export class LightPushCore extends BaseProtocol implements IBaseProtocolCore {
success: null, success: null,
failure: { failure: {
error: ProtocolError.DECODE_FAILED, error: ProtocolError.DECODE_FAILED,
peerId: peer.id peerId: peerId
} }
}; };
} }
@ -156,7 +155,7 @@ export class LightPushCore extends BaseProtocol implements IBaseProtocolCore {
success: null, success: null,
failure: { failure: {
error: ProtocolError.NO_RESPONSE, error: ProtocolError.NO_RESPONSE,
peerId: peer.id peerId: peerId
} }
}; };
} }
@ -168,7 +167,7 @@ export class LightPushCore extends BaseProtocol implements IBaseProtocolCore {
success: null, success: null,
failure: { failure: {
error: rlnErrorCase, error: rlnErrorCase,
peerId: peer.id peerId: peerId
} }
}; };
} }
@ -179,11 +178,11 @@ export class LightPushCore extends BaseProtocol implements IBaseProtocolCore {
success: null, success: null,
failure: { failure: {
error: ProtocolError.REMOTE_PEER_REJECTED, error: ProtocolError.REMOTE_PEER_REJECTED,
peerId: peer.id peerId: peerId
} }
}; };
} }
return { success: peer.id, failure: null }; return { success: peerId, failure: null };
} }
} }

2
packages/core/src/lib/metadata/metadata.ts
View File

@ -55,7 +55,7 @@ class Metadata extends BaseProtocol implements IMetadata {
let stream; let stream;
try { try {
stream = await this.getStream(peer); stream = await this.getStream(peerId);
} catch (error) { } catch (error) {
log.error("Failed to get stream", error); log.error("Failed to get stream", error);
return { return {

6
packages/core/src/lib/store/store.ts
View File

@ -1,4 +1,4 @@
import type { Peer } from "@libp2p/interface"; import type { PeerId } from "@libp2p/interface";
import { import {
IDecodedMessage, IDecodedMessage,
IDecoder, IDecoder,
@ -38,7 +38,7 @@ export class StoreCore extends BaseProtocol implements IStoreCore {
public async *queryPerPage<T extends IDecodedMessage>( public async *queryPerPage<T extends IDecodedMessage>(
queryOpts: QueryRequestParams, queryOpts: QueryRequestParams,
decoders: Map<string, IDecoder<T>>, decoders: Map<string, IDecoder<T>>,
peer: Peer peerId: PeerId
): AsyncGenerator<Promise<T | undefined>[]> { ): AsyncGenerator<Promise<T | undefined>[]> {
if ( if (
queryOpts.contentTopics.toString() !== queryOpts.contentTopics.toString() !==
@ -58,7 +58,7 @@ export class StoreCore extends BaseProtocol implements IStoreCore {
let stream; let stream;
try { try {
stream = await this.getStream(peer); stream = await this.getStream(peerId);
} catch (e) { } catch (e) {
log.error("Failed to get stream", e); log.error("Failed to get stream", e);
break; break;

10
packages/core/src/lib/stream_manager/stream_manager.spec.ts
View File

@ -36,7 +36,7 @@ describe("StreamManager", () => {
streamManager["getConnections"] = (_peerId: PeerId | undefined) => [con1]; streamManager["getConnections"] = (_peerId: PeerId | undefined) => [con1];
const stream = await streamManager.getStream(mockPeer); const stream = await streamManager.getStream(mockPeer.id);
expect(stream).not.to.be.undefined; expect(stream).not.to.be.undefined;
expect(stream?.id).to.be.eq("1"); expect(stream?.id).to.be.eq("1");
@ -48,7 +48,7 @@ describe("StreamManager", () => {
let error: Error | undefined; let error: Error | undefined;
try { try {
await streamManager.getStream(mockPeer); await streamManager.getStream(mockPeer.id);
} catch (e) { } catch (e) {
error = e as Error; error = e as Error;
} }
@ -76,7 +76,7 @@ describe("StreamManager", () => {
con1.newStream = newStreamSpy; con1.newStream = newStreamSpy;
streamManager["getConnections"] = (_peerId: PeerId | undefined) => [con1]; streamManager["getConnections"] = (_peerId: PeerId | undefined) => [con1];
const stream = await streamManager.getStream(mockPeer); const stream = await streamManager.getStream(mockPeer.id);
expect(stream).not.to.be.undefined; expect(stream).not.to.be.undefined;
expect(stream?.id).to.be.eq("2"); expect(stream?.id).to.be.eq("2");
@ -102,8 +102,8 @@ describe("StreamManager", () => {
streamManager["getConnections"] = (_peerId: PeerId | undefined) => [con1]; streamManager["getConnections"] = (_peerId: PeerId | undefined) => [con1];
const [stream1, stream2] = await Promise.all([ const [stream1, stream2] = await Promise.all([
streamManager.getStream(mockPeer), streamManager.getStream(mockPeer.id),
streamManager.getStream(mockPeer) streamManager.getStream(mockPeer.id)
]); ]);
const expected = ["1", "2"].toString(); const expected = ["1", "2"].toString();

34
packages/core/src/lib/stream_manager/stream_manager.ts
View File

@ -21,39 +21,38 @@ export class StreamManager {
this.addEventListener("peer:update", this.handlePeerUpdateStreamPool); this.addEventListener("peer:update", this.handlePeerUpdateStreamPool);
} }
public async getStream(peer: Peer): Promise<Stream> { public async getStream(peerId: PeerId): Promise<Stream> {
const peerId = peer.id.toString(); const peerIdStr = peerId.toString();
const scheduledStream = this.streamPool.get(peerIdStr);
const scheduledStream = this.streamPool.get(peerId);
if (scheduledStream) { if (scheduledStream) {
this.streamPool.delete(peerId); this.streamPool.delete(peerIdStr);
await scheduledStream; await scheduledStream;
} }
let stream = this.getOpenStreamForCodec(peer.id); let stream = this.getOpenStreamForCodec(peerId);
if (stream) { if (stream) {
this.log.info( this.log.info(
`Found existing stream peerId=${peer.id.toString()} multicodec=${this.multicodec}` `Found existing stream peerId=${peerIdStr} multicodec=${this.multicodec}`
); );
this.lockStream(peer.id.toString(), stream); this.lockStream(peerIdStr, stream);
return stream; return stream;
} }
stream = await this.createStream(peer); stream = await this.createStream(peerId);
this.lockStream(peer.id.toString(), stream); this.lockStream(peerIdStr, stream);
return stream; return stream;
} }
private async createStream(peer: Peer, retries = 0): Promise<Stream> { private async createStream(peerId: PeerId, retries = 0): Promise<Stream> {
const connections = this.getConnections(peer.id); const connections = this.getConnections(peerId);
const connection = selectOpenConnection(connections); const connection = selectOpenConnection(connections);
if (!connection) { if (!connection) {
throw new Error( throw new Error(
`Failed to get a connection to the peer peerId=${peer.id.toString()} multicodec=${this.multicodec}` `Failed to get a connection to the peer peerId=${peerId.toString()} multicodec=${this.multicodec}`
); );
} }
@ -63,11 +62,11 @@ export class StreamManager {
for (let i = 0; i < retries + 1; i++) { for (let i = 0; i < retries + 1; i++) {
try { try {
this.log.info( this.log.info(
`Attempting to create a stream for peerId=${peer.id.toString()} multicodec=${this.multicodec}` `Attempting to create a stream for peerId=${peerId.toString()} multicodec=${this.multicodec}`
); );
stream = await connection.newStream(this.multicodec); stream = await connection.newStream(this.multicodec);
this.log.info( this.log.info(
`Created stream for peerId=${peer.id.toString()} multicodec=${this.multicodec}` `Created stream for peerId=${peerId.toString()} multicodec=${this.multicodec}`
); );
break; break;
} catch (error) { } catch (error) {
@ -77,8 +76,7 @@ export class StreamManager {
if (!stream) { if (!stream) {
throw new Error( throw new Error(
`Failed to create a new stream for ${peer.id.toString()} -- ` + `Failed to create a new stream for ${peerId.toString()} -- ` + lastError
lastError
); );
} }
@ -97,7 +95,7 @@ export class StreamManager {
try { try {
this.ongoingCreation.add(peerId); this.ongoingCreation.add(peerId);
await this.createStream(peer); await this.createStream(peer.id);
} catch (error) { } catch (error) {
this.log.error(`Failed to createStreamWithLock:`, error); this.log.error(`Failed to createStreamWithLock:`, error);
} finally { } finally {

2
packages/discovery/src/peer-exchange/waku_peer_exchange.ts
View File

@ -57,7 +57,7 @@ export class WakuPeerExchange extends BaseProtocol implements IPeerExchange {
let stream; let stream;
try { try {
stream = await this.getStream(peer); stream = await this.getStream(peerId);
} catch (err) { } catch (err) {
log.error("Failed to get stream", err); log.error("Failed to get stream", err);
return { return {

8
packages/sdk/src/filter/filter.ts
View File

@ -157,8 +157,8 @@ class Filter implements IFilter {
ensurePubsubTopicIsConfigured(pubsubTopic, this.protocol.pubsubTopics); ensurePubsubTopicIsConfigured(pubsubTopic, this.protocol.pubsubTopics);
const peers = await this.peerManager.getPeers(); const peerIds = await this.peerManager.getPeers();
if (peers.length === 0) { if (peerIds.length === 0) {
return { return {
error: ProtocolError.NO_PEER_AVAILABLE, error: ProtocolError.NO_PEER_AVAILABLE,
subscription: null subscription: null
@ -166,8 +166,8 @@ class Filter implements IFilter {
} }
log.info( log.info(
`Creating filter subscription with ${peers.length} peers: `, `Creating filter subscription with ${peerIds.length} peers: `,
peers.map((peer) => peer.id.toString()) peerIds.map((id) => id.toString())
); );
const subscription = const subscription =

46
packages/sdk/src/filter/subscription_monitor.ts
View File

@ -1,4 +1,4 @@
import type { EventHandler, Peer, PeerId } from "@libp2p/interface"; import type { EventHandler, PeerId } from "@libp2p/interface";
import { FilterCore } from "@waku/core"; import { FilterCore } from "@waku/core";
import type { import type {
FilterProtocolOptions, FilterProtocolOptions,
@ -31,7 +31,7 @@ export class SubscriptionMonitor {
* Cached peers that are in use by subscription. * Cached peers that are in use by subscription.
* Needed to understand if they disconnect later or not. * Needed to understand if they disconnect later or not.
*/ */
public peers: Peer[] = []; public peerIds: PeerId[] = [];
private isStarted: boolean = false; private isStarted: boolean = false;
@ -108,12 +108,12 @@ export class SubscriptionMonitor {
* Method to get peers that are used by particular subscription or, if initially called, peers that can be used by subscription. * Method to get peers that are used by particular subscription or, if initially called, peers that can be used by subscription.
* @returns array of peers * @returns array of peers
*/ */
public async getPeers(): Promise<Peer[]> { public async getPeers(): Promise<PeerId[]> {
if (!this.isStarted) { if (!this.isStarted) {
this.peers = await this.peerManager.getPeers(); this.peerIds = await this.peerManager.getPeers();
} }
return this.peers; return this.peerIds;
} }
/** /**
@ -168,7 +168,7 @@ export class SubscriptionMonitor {
return; return;
} }
await Promise.all(this.peers.map((peer) => this.ping(peer, true))); await Promise.all(this.peerIds.map((id) => this.ping(id, true)));
this.startKeepAlive(); this.startKeepAlive();
} }
@ -178,7 +178,7 @@ export class SubscriptionMonitor {
} }
this.keepAliveIntervalId = setInterval(() => { this.keepAliveIntervalId = setInterval(() => {
void this.peers.map((peer) => this.ping(peer)); void this.peerIds.map((id) => this.ping(id));
}, this.config.keepAliveIntervalMs) as unknown as number; }, this.config.keepAliveIntervalMs) as unknown as number;
} }
@ -216,40 +216,40 @@ export class SubscriptionMonitor {
// this method keeps track of new connections and will trigger subscribe request if needed // this method keeps track of new connections and will trigger subscribe request if needed
private async onPeerConnected(_event: CustomEvent<PeerId>): Promise<void> { private async onPeerConnected(_event: CustomEvent<PeerId>): Promise<void> {
// TODO(weboko): use config.numOfUsedPeers instead of this.peers // TODO(weboko): use config.numOfUsedPeers instead of this.peers
const hasSomePeers = this.peers.length > 0; const hasSomePeers = this.peerIds.length > 0;
if (hasSomePeers) { if (hasSomePeers) {
return; return;
} }
this.peers = await this.peerManager.getPeers(); this.peerIds = await this.peerManager.getPeers();
await Promise.all(this.peers.map((peer) => this.subscribe(peer))); await Promise.all(this.peerIds.map((id) => this.subscribe(id)));
} }
// this method keeps track of disconnects and will trigger subscribe request if needed // this method keeps track of disconnects and will trigger subscribe request if needed
private async onPeerDisconnected(event: CustomEvent<PeerId>): Promise<void> { private async onPeerDisconnected(event: CustomEvent<PeerId>): Promise<void> {
const hasNotBeenUsed = !this.peers.find((p) => p.id.equals(event.detail)); const hasNotBeenUsed = !this.peerIds.find((id) => id.equals(event.detail));
if (hasNotBeenUsed) { if (hasNotBeenUsed) {
return; return;
} }
this.peers = await this.peerManager.getPeers(); this.peerIds = await this.peerManager.getPeers();
// we trigger subscribe for peer that was used before // we trigger subscribe for peer that was used before
// it will expectedly fail and we will initiate addition of a new peer // it will expectedly fail and we will initiate addition of a new peer
await Promise.all(this.peers.map((peer) => this.subscribe(peer))); await Promise.all(this.peerIds.map((id) => this.subscribe(id)));
} }
private async subscribe(_peer: Peer | undefined): Promise<void> { private async subscribe(_peerId: PeerId | undefined): Promise<void> {
let peer: Peer | undefined = _peer; let peerId: PeerId | undefined = _peerId;
for (let i = 0; i < MAX_SUBSCRIBE_ATTEMPTS; i++) { for (let i = 0; i < MAX_SUBSCRIBE_ATTEMPTS; i++) {
if (!peer) { if (!peerId) {
return; return;
} }
const response = await this.filter.subscribe( const response = await this.filter.subscribe(
this.pubsubTopic, this.pubsubTopic,
peer, peerId,
Array.from(this.activeSubscriptions.keys()) Array.from(this.activeSubscriptions.keys())
); );
@ -257,19 +257,19 @@ export class SubscriptionMonitor {
return; return;
} }
peer = await this.peerManager.requestRenew(peer.id); peerId = await this.peerManager.requestRenew(peerId);
} }
} }
private async ping( private async ping(
peer: Peer, peerId: PeerId,
renewOnFirstFail: boolean = false renewOnFirstFail: boolean = false
): Promise<void> { ): Promise<void> {
const peerIdStr = peer.id.toString(); const peerIdStr = peerId.toString();
const response = await this.filter.ping(peer); const response = await this.filter.ping(peerId);
if (response.failure && renewOnFirstFail) { if (response.failure && renewOnFirstFail) {
const newPeer = await this.peerManager.requestRenew(peer.id); const newPeer = await this.peerManager.requestRenew(peerId);
await this.subscribe(newPeer); await this.subscribe(newPeer);
return; return;
} }
@ -286,7 +286,7 @@ export class SubscriptionMonitor {
const madeAttempts = this.pingFailedAttempts.get(peerIdStr) || 0; const madeAttempts = this.pingFailedAttempts.get(peerIdStr) || 0;
if (madeAttempts >= this.config.pingsBeforePeerRenewed) { if (madeAttempts >= this.config.pingsBeforePeerRenewed) {
const newPeer = await this.peerManager.requestRenew(peer.id); const newPeer = await this.peerManager.requestRenew(peerId);
await this.subscribe(newPeer); await this.subscribe(newPeer);
} }
} }

19
packages/sdk/src/light_push/light_push.spec.ts
View File

@ -1,4 +1,4 @@
import { Peer } from "@libp2p/interface"; import { Peer, PeerId } from "@libp2p/interface";
import { import {
ConnectionManager, ConnectionManager,
createEncoder, createEncoder,
@ -59,8 +59,8 @@ describe("LightPush SDK", () => {
lightPush = mockLightPush({ libp2p, numPeersToUse: 2 }); lightPush = mockLightPush({ libp2p, numPeersToUse: 2 });
let sendSpy = sinon.spy( let sendSpy = sinon.spy(
(_encoder: any, _message: any, peer: Peer) => (_encoder: any, _message: any, peerId: PeerId) =>
({ success: peer.id }) as any ({ success: peerId }) as any
); );
lightPush.protocol.send = sendSpy; lightPush.protocol.send = sendSpy;
@ -74,8 +74,8 @@ describe("LightPush SDK", () => {
// check if setting another value works // check if setting another value works
lightPush = mockLightPush({ libp2p, numPeersToUse: 3 }); lightPush = mockLightPush({ libp2p, numPeersToUse: 3 });
sendSpy = sinon.spy( sendSpy = sinon.spy(
(_encoder: any, _message: any, peer: Peer) => (_encoder: any, _message: any, peerId: PeerId) =>
({ success: peer.id }) as any ({ success: peerId }) as any
); );
lightPush.protocol.send = sendSpy; lightPush.protocol.send = sendSpy;
@ -91,9 +91,9 @@ describe("LightPush SDK", () => {
}); });
lightPush = mockLightPush({ libp2p }); lightPush = mockLightPush({ libp2p });
let sendSpy = sinon.spy((_encoder: any, _message: any, peer: Peer) => { let sendSpy = sinon.spy((_encoder: any, _message: any, peerId: PeerId) => {
if (peer.id.toString() === "1") { if (peerId.toString() === "1") {
return { success: peer.id }; return { success: peerId };
} }
return { failure: { error: "problem" } }; return { failure: { error: "problem" } };
@ -108,7 +108,7 @@ describe("LightPush SDK", () => {
{ autoRetry: true } { autoRetry: true }
); );
expect(attemptRetriesSpy.calledOnce).to.be.true; expect(attemptRetriesSpy.callCount).to.be.eq(1);
expect(result.successes?.length).to.be.eq(1); expect(result.successes?.length).to.be.eq(1);
expect(result.failures?.length).to.be.eq(1); expect(result.failures?.length).to.be.eq(1);
@ -162,7 +162,6 @@ function mockLightPush(options: MockLightPushOptions): LightPush {
getPeers: () => getPeers: () =>
options.libp2p options.libp2p
.getPeers() .getPeers()
.map((id) => mockPeer(id.toString()))
.slice(0, options.numPeersToUse || options.libp2p.getPeers().length) .slice(0, options.numPeersToUse || options.libp2p.getPeers().length)
} as unknown as PeerManager, } as unknown as PeerManager,
options.libp2p options.libp2p

22
packages/sdk/src/light_push/light_push.ts
View File

@ -1,4 +1,4 @@
import type { Peer, PeerId } from "@libp2p/interface"; import type { PeerId } from "@libp2p/interface";
import { ConnectionManager, getHealthManager, LightPushCore } from "@waku/core"; import { ConnectionManager, getHealthManager, LightPushCore } from "@waku/core";
import { import {
type CoreProtocolResult, type CoreProtocolResult,
@ -23,7 +23,7 @@ const DEFAULT_SEND_OPTIONS: ISenderOptions = {
maxAttempts: DEFAULT_MAX_ATTEMPTS maxAttempts: DEFAULT_MAX_ATTEMPTS
}; };
type RetryCallback = (peer: Peer) => Promise<CoreProtocolResult>; type RetryCallback = (peerId: PeerId) => Promise<CoreProtocolResult>;
export class LightPush implements ILightPush { export class LightPush implements ILightPush {
public readonly protocol: LightPushCore; public readonly protocol: LightPushCore;
@ -59,8 +59,8 @@ export class LightPush implements ILightPush {
}; };
} }
const peers = await this.peerManager.getPeers(); const peerIds = await this.peerManager.getPeers();
if (peers.length === 0) { if (peerIds.length === 0) {
return { return {
successes, successes,
failures: [ failures: [
@ -72,7 +72,7 @@ export class LightPush implements ILightPush {
} }
const results = await Promise.allSettled( const results = await Promise.allSettled(
peers.map((peer) => this.protocol.send(encoder, message, peer)) peerIds.map((id) => this.protocol.send(encoder, message, id))
); );
for (const result of results) { for (const result of results) {
@ -94,7 +94,7 @@ export class LightPush implements ILightPush {
if (options?.autoRetry) { if (options?.autoRetry) {
void this.attemptRetries( void this.attemptRetries(
(peer: Peer) => this.protocol.send(encoder, message, peer), (id: PeerId) => this.protocol.send(encoder, message, id),
options.maxAttempts options.maxAttempts
); );
} }
@ -117,23 +117,23 @@ export class LightPush implements ILightPush {
maxAttempts?: number maxAttempts?: number
): Promise<void> { ): Promise<void> {
maxAttempts = maxAttempts || DEFAULT_MAX_ATTEMPTS; maxAttempts = maxAttempts || DEFAULT_MAX_ATTEMPTS;
const connectedPeers = await this.peerManager.getPeers(); const peerIds = await this.peerManager.getPeers();
if (connectedPeers.length === 0) { if (peerIds.length === 0) {
log.warn("Cannot retry with no connected peers."); log.warn("Cannot retry with no connected peers.");
return; return;
} }
for (let i = 0; i < maxAttempts; i++) { for (let i = 0; i < maxAttempts; i++) {
const peer = connectedPeers[i % connectedPeers.length]; // always present as we checked for the length already const id = peerIds[i % peerIds.length]; // always present as we checked for the length already
const response = await fn(peer); const response = await fn(id);
if (response.success) { if (response.success) {
return; return;
} }
log.info( log.info(
`Attempted retry for peer:${peer.id} failed with:${response?.failure?.error}` `Attempted retry for peer:${id} failed with:${response?.failure?.error}`
); );
} }
} }

6
packages/sdk/src/peer_manager/peer_manager.spec.ts
View File

@ -48,9 +48,9 @@ describe("PeerManager", () => {
]; ];
sinon.stub(libp2p, "getConnections").returns(connections); sinon.stub(libp2p, "getConnections").returns(connections);
const peer = await peerManager.requestRenew("1"); const peerId = await peerManager.requestRenew("1");
expect(peer).to.not.be.undefined; expect(peerId).to.not.be.undefined;
expect(peer?.id).to.not.equal("1"); expect(peerId).to.not.equal("1");
}); });
it("should handle connection events", () => { it("should handle connection events", () => {

25
packages/sdk/src/peer_manager/peer_manager.ts
View File

@ -1,4 +1,4 @@
import { Connection, Peer, PeerId } from "@libp2p/interface"; import { Connection, PeerId } from "@libp2p/interface";
import { Libp2p } from "@waku/interfaces"; import { Libp2p } from "@waku/interfaces";
import { Logger } from "@waku/utils"; import { Logger } from "@waku/utils";
@ -37,15 +37,13 @@ export class PeerManager {
this.stopConnectionListener(); this.stopConnectionListener();
} }
public async getPeers(): Promise<Peer[]> { public async getPeers(): Promise<PeerId[]> {
return Promise.all( return Promise.all(this.getLockedConnections().map((c) => c.remotePeer));
this.getLockedConnections().map((c) => this.mapConnectionToPeer(c))
);
} }
public async requestRenew( public async requestRenew(
peerId: PeerId | string peerId: PeerId | string
): Promise<Peer | undefined> { ): Promise<PeerId | undefined> {
const lockedConnections = this.getLockedConnections(); const lockedConnections = this.getLockedConnections();
const neededPeers = this.numPeersToUse - lockedConnections.length; const neededPeers = this.numPeersToUse - lockedConnections.length;
@ -58,12 +56,12 @@ export class PeerManager {
.filter((c) => !c.remotePeer.equals(peerId)) .filter((c) => !c.remotePeer.equals(peerId))
.slice(0, neededPeers) .slice(0, neededPeers)
.map((c) => this.lockConnection(c)) .map((c) => this.lockConnection(c))
.map((c) => this.mapConnectionToPeer(c)) .map((c) => c.remotePeer)
); );
const newPeer = result[0]; const newPeerId = result[0];
if (!newPeer) { if (!newPeerId) {
log.warn( log.warn(
`requestRenew: Couldn't renew peer ${peerId.toString()} - no peers.` `requestRenew: Couldn't renew peer ${peerId.toString()} - no peers.`
); );
@ -71,10 +69,10 @@ export class PeerManager {
} }
log.info( log.info(
`requestRenew: Renewed peer ${peerId.toString()} to ${newPeer.id.toString()}` `requestRenew: Renewed peer ${peerId.toString()} to ${newPeerId.toString()}`
); );
return newPeer; return newPeerId;
} }
private startConnectionListener(): void { private startConnectionListener(): void {
@ -133,9 +131,4 @@ export class PeerManager {
private isConnectionLocked(c: Connection): boolean { private isConnectionLocked(c: Connection): boolean {
return c.tags.includes(CONNECTION_LOCK_TAG); return c.tags.includes(CONNECTION_LOCK_TAG);
} }
private async mapConnectionToPeer(c: Connection): Promise<Peer> {
const peerId = c.remotePeer;
return this.libp2p.peerStore.get(peerId);
}
} }

19
packages/sdk/src/store/store.ts
View File

@ -1,4 +1,4 @@
import type { Peer } from "@libp2p/interface"; import type { PeerId } from "@libp2p/interface";
import { ConnectionManager, StoreCore } from "@waku/core"; import { ConnectionManager, StoreCore } from "@waku/core";
import { import {
IDecodedMessage, IDecodedMessage,
@ -223,26 +223,29 @@ export class Store implements IStore {
}; };
} }
private async getPeerToUse(): Promise<Peer | undefined> { private async getPeerToUse(): Promise<PeerId | undefined> {
let peer: Peer | undefined; let peerId: PeerId | undefined;
if (this.options?.peer) { if (this.options?.peer) {
const connectedPeers = await this.connectionManager.getConnectedPeers(); const connectedPeers = await this.connectionManager.getConnectedPeers();
peer = connectedPeers.find((p) => p.id.toString() === this.options?.peer); const peer = connectedPeers.find(
(p) => p.id.toString() === this.options?.peer
);
peerId = peer?.id;
if (!peer) { if (!peerId) {
log.warn( log.warn(
`Passed node to use for Store not found: ${this.options.peer}. Attempting to use random peers.` `Passed node to use for Store not found: ${this.options.peer}. Attempting to use random peers.`
); );
} }
} }
const peers = await this.peerManager.getPeers(); const peerIds = await this.peerManager.getPeers();
if (peers.length > 0) { if (peerIds.length > 0) {
// TODO(weboko): implement smart way of getting a peer https://github.com/waku-org/js-waku/issues/2243 // TODO(weboko): implement smart way of getting a peer https://github.com/waku-org/js-waku/issues/2243
return peers[Math.floor(Math.random() * peers.length)]; return peerIds[Math.floor(Math.random() * peerIds.length)];
} }
log.error("No peers available to use."); log.error("No peers available to use.");