From dacf2875b7c2984eeb83a6562205324b4bc3fe80 Mon Sep 17 00:00:00 2001 From: Julien Date: Thu, 2 Mar 2023 22:23:25 -0800 Subject: [PATCH] General WsProvider extensibility improvements (#5467) * Make endpoints index selection extensible * Expose endpoint * Ensure events happen last * Added endpoint stats --- packages/rpc-provider/src/types.ts | 20 ++++++----- packages/rpc-provider/src/ws/index.ts | 51 +++++++++++++++++++++------ 2 files changed, 52 insertions(+), 19 deletions(-) diff --git a/packages/rpc-provider/src/types.ts b/packages/rpc-provider/src/types.ts index a98ad46eb1..2ab35b28f4 100644 --- a/packages/rpc-provider/src/types.ts +++ b/packages/rpc-provider/src/types.ts @@ -67,18 +67,20 @@ export interface ProviderInterface { unsubscribe (type: string, method: string, id: number | string): Promise; } +export interface EndpointStats { + bytesRecv: number; + bytesSent: number; + cached: number; + errors: number; + requests: number; + subscriptions: number; + timeout: number; +} + export interface ProviderStats { active: { requests: number; subscriptions: number; }; - total: { - bytesRecv: number; - bytesSent: number; - cached: number; - errors: number; - requests: number; - subscriptions: number; - timeout: number; - }; + total: EndpointStats; } diff --git a/packages/rpc-provider/src/ws/index.ts b/packages/rpc-provider/src/ws/index.ts index 26d874ed45..fca065ad88 100644 --- a/packages/rpc-provider/src/ws/index.ts +++ b/packages/rpc-provider/src/ws/index.ts @@ -3,7 +3,7 @@ /* eslint-disable camelcase */ -import type { JsonRpcResponse, ProviderInterface, ProviderInterfaceCallback, ProviderInterfaceEmitCb, ProviderInterfaceEmitted, ProviderStats } from '../types'; +import type { EndpointStats, JsonRpcResponse, ProviderInterface, ProviderInterfaceCallback, ProviderInterfaceEmitCb, ProviderInterfaceEmitted, ProviderStats } from '../types'; import EventEmitter from 'eventemitter3'; @@ -59,6 +59,10 @@ function eraseRecord (record: Record, cb?: (item: T) => void): voi }); } +function defaultEndpointStats (): EndpointStats { + return { bytesRecv: 0, bytesSent: 0, cached: 0, errors: 0, requests: 0, subscriptions: 0, timeout: 0 }; +} + /** * # @polkadot/rpc-provider/ws * @@ -92,6 +96,7 @@ export class WsProvider implements ProviderInterface { #autoConnectMs: number; #endpointIndex: number; + #endpointStats: EndpointStats; #isConnected = false; #subscriptions: Record = {}; #timeoutId?: ReturnType | null = null; @@ -128,8 +133,9 @@ export class WsProvider implements ProviderInterface { this.#websocket = null; this.#stats = { active: { requests: 0, subscriptions: 0 }, - total: { bytesRecv: 0, bytesSent: 0, cached: 0, errors: 0, requests: 0, subscriptions: 0, timeout: 0 } + total: defaultEndpointStats() }; + this.#endpointStats = defaultEndpointStats(); this.#timeout = timeout || DEFAULT_TIMEOUT_MS; if (autoConnectMs > 0) { @@ -174,6 +180,10 @@ export class WsProvider implements ProviderInterface { return this.#isReadyPromise; } + public get endpoint (): string { + return this.#endpoints[this.#endpointIndex]; + } + /** * @description Returns a clone of the object */ @@ -181,6 +191,10 @@ export class WsProvider implements ProviderInterface { return new WsProvider(this.#endpoints); } + protected selectEndpointIndex (endpoints: string[]): number { + return (this.#endpointIndex + 1) % endpoints.length; + } + /** * @summary Manually connect * @description The [[WsProvider]] connects automatically by default, however if you decided otherwise, you may @@ -193,14 +207,14 @@ export class WsProvider implements ProviderInterface { } try { - this.#endpointIndex = (this.#endpointIndex + 1) % this.#endpoints.length; + this.#endpointIndex = this.selectEndpointIndex(this.#endpoints); // the as typeof WebSocket here is Deno-specific - not available on the globalThis this.#websocket = typeof xglobal.WebSocket !== 'undefined' && isChildClass(xglobal.WebSocket as typeof WebSocket, WebSocket) - ? new WebSocket(this.#endpoints[this.#endpointIndex]) + ? new WebSocket(this.endpoint) // eslint-disable-next-line @typescript-eslint/ban-ts-comment // @ts-ignore - WS may be an instance of w3cwebsocket, which supports headers - : new WebSocket(this.#endpoints[this.#endpointIndex], undefined, undefined, this.#headers, undefined, { + : new WebSocket(this.endpoint, undefined, undefined, this.#headers, undefined, { // default: true fragmentOutgoingMessages: true, // default: 16K (bump, the Node has issues with too many fragments, e.g. on setCode) @@ -281,6 +295,10 @@ export class WsProvider implements ProviderInterface { }; } + public get endpointStats (): EndpointStats { + return this.#endpointStats; + } + /** * @summary Listens on events after having subscribed using the [[subscribe]] function. * @param {ProviderInterfaceEmitted} type Event @@ -302,6 +320,7 @@ export class WsProvider implements ProviderInterface { * @param subscription Subscription details (internally used) */ public send (method: string, params: unknown[], isCacheable?: boolean, subscription?: SubscriptionHandler): Promise { + this.#endpointStats.requests++; this.#stats.total.requests++; const [id, body] = this.#coder.encodeJson(method, params); @@ -316,6 +335,7 @@ export class WsProvider implements ProviderInterface { this.#callCache.set(body, resultPromise); } } else { + this.#endpointStats.cached++; this.#stats.total.cached++; } @@ -344,9 +364,11 @@ export class WsProvider implements ProviderInterface { start: Date.now(), subscription }; + this.#endpointStats.bytesSent += body.length; this.#stats.total.bytesSent += body.length; this.#websocket.send(body); } catch (error) { + this.#endpointStats.errors++; this.#stats.total.errors++; reject(error); @@ -373,6 +395,7 @@ export class WsProvider implements ProviderInterface { * ``` */ public subscribe (type: string, method: string, params: unknown[], callback: ProviderInterfaceCallback): Promise { + this.#endpointStats.subscriptions++; this.#stats.total.subscriptions++; // subscriptions are not cached, LRU applies to .at() only @@ -411,7 +434,7 @@ export class WsProvider implements ProviderInterface { }; #onSocketClose = (event: CloseEvent): void => { - const error = new Error(`disconnected from ${this.#endpoints[this.#endpointIndex]}: ${event.code}:: ${event.reason || getWSErrorString(event.code)}`); + const error = new Error(`disconnected from ${this.endpoint}: ${event.code}:: ${event.reason || getWSErrorString(event.code)}`); if (this.#autoConnectMs > 0) { l.error(error.message); @@ -432,8 +455,6 @@ export class WsProvider implements ProviderInterface { this.#timeoutId = null; } - this.#emit('disconnected'); - // reject all hanging requests eraseRecord(this.#handlers, (h) => { try { @@ -445,6 +466,11 @@ export class WsProvider implements ProviderInterface { }); eraseRecord(this.#waitingForId); + // Reset stats for active endpoint + this.#endpointStats = defaultEndpointStats(); + + this.#emit('disconnected'); + if (this.#autoConnectMs > 0) { setTimeout((): void => { this.connectWithRetry().catch(() => { @@ -462,6 +488,7 @@ export class WsProvider implements ProviderInterface { #onSocketMessage = (message: MessageEvent): void => { l.debug(() => ['received', message.data]); + this.#endpointStats.bytesRecv += message.data.length; this.#stats.total.bytesRecv += message.data.length; const response = JSON.parse(message.data) as JsonRpcResponse; @@ -502,6 +529,7 @@ export class WsProvider implements ProviderInterface { } } } catch (error) { + this.#endpointStats.errors++; this.#stats.total.errors++; handler.callback(error as Error, undefined); @@ -532,6 +560,7 @@ export class WsProvider implements ProviderInterface { handler.callback(null, result); } catch (error) { + this.#endpointStats.errors++; this.#stats.total.errors++; handler.callback(error as Error, undefined); @@ -543,13 +572,14 @@ export class WsProvider implements ProviderInterface { throw new Error('WebSocket cannot be null in onOpen'); } - l.debug(() => ['connected to', this.#endpoints[this.#endpointIndex]]); + l.debug(() => ['connected to', this.endpoint]); this.#isConnected = true; - this.#emit('connected'); this.#resubscribe(); + this.#emit('connected'); + return true; }; @@ -590,6 +620,7 @@ export class WsProvider implements ProviderInterface { // ignore } + this.#endpointStats.timeout++; this.#stats.total.timeout++; delete this.#handlers[ids[i]]; }