Merge pull request #118 from lidofinance/feat/provider-rpc-events

Add RPC events and middleware data for better monitoring
lidofinance · Sep 12, 2024 · 14b9d8b · 14b9d8b
2 parents 3df4406 + 8a4dd7e
commit 14b9d8b
Show file tree

Hide file tree

Showing 11 changed files with 352 additions and 327 deletions.
diff --git a/packages/execution/src/common/networks.ts b/packages/execution/src/common/networks.ts
@@ -1,5 +1,14 @@
 import { Network } from '@ethersproject/networks';
 import { Networkish } from '../interfaces/networkish';
+import { ConnectionInfo } from '@ethersproject/web';
+
+const IP_V4_REGEX = new RegExp(
+  /^(?<domain>\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})(?::(?<port>\d+))?/i,
+);
+
+const DOMAIN_REGEX = new RegExp(
+  /^(?<protocol>https?:\/\/)(?=(?<fqdn>[^:/]+))(?:(?<service>www|ww\d|cdn|ftp|mail|pop\d?|ns\d?|git)\.)?(?:(?<subdomain>[^:/]+)\.)*(?<domain>[^:/]+\.[a-z0-9]+)(?::(?<port>\d+))?(?<path>\/[^?]*)?(?:\?(?<query>[^#]*))?(?:#(?<hash>.*))?/i,
+);
 
 export const networksEqual = (
   networkA: Network,
@@ -22,3 +31,22 @@ export const networksChainsEqual = (
   networkA: Network,
   networkB: Networkish,
 ): boolean => networkA.chainId === getNetworkChain(networkB);
+
+export const getConnectionFQDN = (
+  connectionInfo: ConnectionInfo | string,
+): string => {
+  const urlLike =
+    typeof connectionInfo === 'string' ? connectionInfo : connectionInfo.url;
+
+  const ipGroups = urlLike.match(IP_V4_REGEX)?.groups;
+
+  if (ipGroups) {
+    /* istanbul ignore next */
+    return ipGroups.domain ?? '';
+  }
+
+  const groups = urlLike.match(DOMAIN_REGEX)?.groups;
+
+  /* istanbul ignore next */
+  return groups?.fqdn ?? '';
+};
diff --git a/packages/execution/src/events/index.ts b/packages/execution/src/events/index.ts
@@ -0,0 +1,62 @@
+import { SimpleFallbackJsonRpcBatchProvider } from '../provider/simple-fallback-json-rpc-batch-provider';
+import { AllProvidersFailedError } from '../error';
+import {
+  ExtendedJsonRpcBatchProvider,
+  JsonRpcRequest,
+  JsonRpcResponse,
+} from '../provider/extended-json-rpc-batch-provider';
+
+export type FallbackProviderRequestFailedAllEvent = {
+  action: 'fallback-provider:request:failed:all';
+  provider: SimpleFallbackJsonRpcBatchProvider;
+  error: AllProvidersFailedError;
+};
+
+export type FallbackProviderRequestNonRetryableErrorEvent = {
+  action: 'fallback-provider:request:non-retryable-error';
+  provider: SimpleFallbackJsonRpcBatchProvider;
+  error: Error | unknown;
+};
+
+export type FallbackProviderRequestEvent = {
+  action: 'fallback-provider:request';
+  provider: SimpleFallbackJsonRpcBatchProvider;
+  activeFallbackProviderIndex: number;
+  fallbackProvidersCount: number;
+  domain: string;
+  retryAttempt: number;
+};
+
+export type ProviderResponseBatchedErrorEvent = {
+  action: 'provider:response-batched:error';
+  error: Error;
+  request: JsonRpcRequest[];
+  provider: ExtendedJsonRpcBatchProvider;
+  domain: string;
+};
+
+export type ProviderResponseBatchedEvent = {
+  action: 'provider:response-batched';
+  request: JsonRpcRequest[];
+  response: JsonRpcResponse[] | JsonRpcResponse;
+  provider: ExtendedJsonRpcBatchProvider;
+  domain: string;
+};
+
+export type ProviderRequestBatchedEvent = {
+  action: 'provider:request-batched';
+  request: JsonRpcRequest[];
+  provider: ExtendedJsonRpcBatchProvider;
+  domain: string;
+};
+
+export type ProviderEvents =
+  | ProviderRequestBatchedEvent
+  | ProviderResponseBatchedEvent
+  | ProviderResponseBatchedErrorEvent;
+
+export type FallbackProviderEvents =
+  | ProviderEvents
+  | FallbackProviderRequestEvent
+  | FallbackProviderRequestNonRetryableErrorEvent
+  | FallbackProviderRequestFailedAllEvent;
diff --git a/packages/execution/src/index.ts b/packages/execution/src/index.ts
@@ -9,3 +9,4 @@ export * from './interfaces/module.options';
 export * from './interfaces/non-empty-array';
 export * from './ethers/fee-history';
 export * from './error';
+export * from './events';
diff --git a/packages/execution/src/provider/extended-json-rpc-batch-provider.ts b/packages/execution/src/provider/extended-json-rpc-batch-provider.ts
@@ -21,6 +21,25 @@ import { FeeHistory, getFeeHistory } from '../ethers/fee-history';
 import { ErrorCode } from '../error/codes/error-codes';
 import { TraceConfig, TraceResult } from '../interfaces/debug-traces';
 import { getDebugTraceBlockByHash } from '../ethers/debug-trace-block-by-hash';
+import { getConnectionFQDN } from '../common/networks';
+import { EventEmitter } from 'events';
+import {
+  ProviderEvents,
+  ProviderRequestBatchedEvent,
+  ProviderResponseBatchedErrorEvent,
+  ProviderResponseBatchedEvent,
+} from '../events';
+
+// this will help with autocomplete
+export interface ExtendedJsonRpcBatchProviderEventEmitter
+  extends NodeJS.EventEmitter {
+  on(eventName: 'rpc', listener: (event: ProviderEvents) => void): this;
+  once(eventName: 'rpc', listener: (event: ProviderEvents) => void): this;
+  addListener(
+    eventName: 'rpc',
+    listener: (event: ProviderEvents) => void,
+  ): this;
+}
 
 export interface RequestPolicy {
   jsonRpcMaxBatchSize: number;
@@ -104,6 +123,8 @@ export class ExtendedJsonRpcBatchProvider extends JsonRpcProvider {
   protected _concurrencyLimiter: LimitFunction;
   protected _tickCounter = 0;
   protected _fetchMiddlewareService: MiddlewareService<Promise<any>>;
+  protected _domain: string;
+  protected _eventEmitter: ExtendedJsonRpcBatchProviderEventEmitter;
 
   public constructor(
     url: ConnectionInfo | string,
@@ -112,6 +133,8 @@ export class ExtendedJsonRpcBatchProvider extends JsonRpcProvider {
     fetchMiddlewares: MiddlewareCallback<Promise<any>>[] = [],
   ) {
     super(url, network);
+    this._eventEmitter = new EventEmitter();
+    this._domain = getConnectionFQDN(url);
     this._requestPolicy = requestPolicy ?? {
       jsonRpcMaxBatchSize: 200,
       maxConcurrentRequests: 5,
@@ -152,28 +175,33 @@ export class ExtendedJsonRpcBatchProvider extends JsonRpcProvider {
 
       const batchRequest = batch.map((intent) => intent.request);
 
-      this.emit('debug', {
-        action: 'requestBatch',
+      const event: ProviderRequestBatchedEvent = {
+        action: 'provider:request-batched',
         request: deepCopy(batchRequest),
         provider: this,
-      });
+        domain: this._domain,
+      };
+      this._eventEmitter.emit('rpc', event);
 
       this._concurrencyLimiter(() => {
         return this._fetchMiddlewareService.go(
           () => this.fetchJson(this.connection, JSON.stringify(batchRequest)),
           {
             provider: this,
+            domain: this._domain,
           },
         );
       })
         .then(
           (batchResult: JsonRpcResponse[] | JsonRpcResponse) => {
-            this.emit('debug', {
-              action: 'response',
+            const event: ProviderResponseBatchedEvent = {
+              action: 'provider:response-batched',
               request: deepCopy(batchRequest),
               response: deepCopy(batchResult),
               provider: this,
-            });
+              domain: this._domain,
+            };
+            this._eventEmitter.emit('rpc', event);
 
             if (!Array.isArray(batchResult)) {
               const errMessage = 'Unexpected batch result.';
@@ -216,12 +244,14 @@ export class ExtendedJsonRpcBatchProvider extends JsonRpcProvider {
             });
           },
           (error: Error) => {
-            this.emit('debug', {
-              action: 'response',
+            const event: ProviderResponseBatchedErrorEvent = {
+              action: 'provider:response-batched:error',
               error: error,
               request: deepCopy(batchRequest),
               provider: this,
-            });
+              domain: this._domain,
+            };
+            this._eventEmitter.emit('rpc', event);
 
             batch.forEach((inflightRequest) => {
               inflightRequest.reject(error);
@@ -230,12 +260,14 @@ export class ExtendedJsonRpcBatchProvider extends JsonRpcProvider {
         )
         .catch((error: Error) => {
           // catch errors happening in the 'then' callback
-          this.emit('debug', {
-            action: 'response',
+          const event: ProviderResponseBatchedErrorEvent = {
+            action: 'provider:response-batched:error',
             error: error,
             request: deepCopy(batchRequest),
             provider: this,
-          });
+            domain: this._domain,
+          };
+          this._eventEmitter.emit('rpc', event);
 
           batch.forEach((inflightRequest) => {
             inflightRequest.reject(error);
@@ -299,6 +331,14 @@ export class ExtendedJsonRpcBatchProvider extends JsonRpcProvider {
     this._fetchMiddlewareService.use(callback);
   }
 
+  public get domain(): string {
+    return this._domain;
+  }
+
+  public get eventEmitter() {
+    return this._eventEmitter;
+  }
+
   public send(method: string, params: Array<unknown>): Promise<unknown> {
     const request: JsonRpcRequest = {
       method: method,

diff --git a/packages/execution/src/provider/simple-fallback-json-rpc-batch-provider.ts b/packages/execution/src/provider/simple-fallback-json-rpc-batch-provider.ts
@@ -27,6 +27,13 @@ import { AllProvidersFailedError } from '../error/all-providers-failed.error';
 import { FeeHistory, getFeeHistory } from '../ethers/fee-history';
 import { TraceConfig, TraceResult } from '../interfaces/debug-traces';
 import { getDebugTraceBlockByHash } from '../ethers/debug-trace-block-by-hash';
+import { EventEmitter } from 'events';
+import {
+  FallbackProviderEvents,
+  FallbackProviderRequestEvent,
+  FallbackProviderRequestFailedAllEvent,
+  FallbackProviderRequestNonRetryableErrorEvent,
+} from '../events';
 
 /**
  * EIP-1898 support
@@ -62,6 +69,20 @@ declare module '@ethersproject/providers' {
   }
 }
 
+// this will help with autocomplete
+export interface SimpleFallbackJsonRpcBatchProviderEventEmitter
+  extends NodeJS.EventEmitter {
+  on(eventName: 'rpc', listener: (event: FallbackProviderEvents) => void): this;
+  once(
+    eventName: 'rpc',
+    listener: (event: FallbackProviderEvents) => void,
+  ): this;
+  addListener(
+    eventName: 'rpc',
+    listener: (event: FallbackProviderEvents) => void,
+  ): this;
+}
+
 @Injectable()
 export class SimpleFallbackJsonRpcBatchProvider extends BaseProvider {
   protected config: SimpleFallbackProviderConfig;
@@ -73,12 +94,14 @@ export class SimpleFallbackJsonRpcBatchProvider extends BaseProvider {
   // it is crucial not to mix these two errors
   protected lastPerformError: Error | null | unknown = null; // last error for 'perform' operations, is batch-oriented
   protected lastError: Error | null | unknown = null; // last error for whole provider
+  protected _eventEmitter: SimpleFallbackJsonRpcBatchProviderEventEmitter;
 
   public constructor(
     config: SimpleFallbackProviderConfig,
     logger: LoggerService,
   ) {
     super(config.network);
+    this._eventEmitter = new EventEmitter();
     this.config = {
       maxRetries: 3,
       minBackoffMs: 500,
@@ -113,6 +136,12 @@ export class SimpleFallbackJsonRpcBatchProvider extends BaseProvider {
         config.requestPolicy,
         config.fetchMiddlewares ?? [],
       );
+
+      // re-emitting events from fallback-providers
+      provider.eventEmitter.on('rpc', (event) => {
+        this._eventEmitter.emit('rpc', event);
+      });
+
       return {
         network: null,
         provider,
@@ -245,16 +274,37 @@ export class SimpleFallbackJsonRpcBatchProvider extends BaseProvider {
     // maximum number of switching is limited to total fallback provider count
     while (attempt < this.fallbackProviders.length) {
       try {
+        let performRetryAttempt = 0;
         attempt++;
         // awaiting is extremely important here
         // without it, the error will not be caught in current try-catch scope
-        return await retry(() =>
-          this.provider.provider.perform(method, params),
-        );
+        return await retry(() => {
+          const provider = this.provider;
+
+          const event: FallbackProviderRequestEvent = {
+            action: 'fallback-provider:request',
+            provider: this,
+            activeFallbackProviderIndex: this.activeFallbackProviderIndex,
+            fallbackProvidersCount: this.fallbackProviders.length,
+            domain: provider.provider.domain,
+            retryAttempt: performRetryAttempt,
+          };
+          this._eventEmitter.emit('rpc', event);
+
+          performRetryAttempt++;
+          return provider.provider.perform(method, params);
+        });
       } catch (e) {
         this.lastError = e;
+
         // checking that error should not be retried on another provider
         if (this.isNonRetryableError(e)) {
+          const event: FallbackProviderRequestNonRetryableErrorEvent = {
+            action: 'fallback-provider:request:non-retryable-error',
+            provider: this,
+            error: e,
+          };
+          this._eventEmitter.emit('rpc', event);
           throw e;
         }
 
@@ -279,6 +329,14 @@ export class SimpleFallbackJsonRpcBatchProvider extends BaseProvider {
       'All attempts to do ETH1 RPC request failed',
     );
     allProvidersFailedError.cause = this.lastError;
+
+    const event: FallbackProviderRequestFailedAllEvent = {
+      action: 'fallback-provider:request:failed:all',
+      provider: this,
+      error: allProvidersFailedError,
+    };
+    this._eventEmitter.emit('rpc', event);
+
     throw allProvidersFailedError;
   }
 
@@ -382,4 +440,8 @@ export class SimpleFallbackJsonRpcBatchProvider extends BaseProvider {
   public get activeProviderIndex() {
     return this.activeFallbackProviderIndex;
   }
+
+  public get eventEmitter() {
+    return this._eventEmitter;
+  }
 }
diff --git a/packages/execution/test/fallback-provider.spec.ts b/packages/execution/test/fallback-provider.spec.ts
@@ -25,7 +25,7 @@ import { MiddlewareCallback } from '@lido-nestjs/middleware';
 import { Network } from '@ethersproject/networks';
 import { nonRetryableErrors } from '../src/common/errors';
 import { ErrorCode, Logger } from '@ethersproject/logger';
-import { AllProvidersFailedError } from '../src';
+import { AllProvidersFailedError, FallbackProviderEvents } from '../src';
 
 export type MockedExtendedJsonRpcBatchProvider =
   ExtendedJsonRpcBatchProvider & {
@@ -797,5 +797,21 @@ describe('Execution module. ', () => {
       // second 'getBlock' fetch call to second provider that does not fail
       expect(mockedFallbackProviderFetch[1]).toBeCalledTimes(2);
     });
+
+    test('should emit `fallback-provider:request` events', async () => {
+      await createMocks(2);
+
+      let retryAttempt = NaN;
+      mockedProvider.eventEmitter.on('rpc', (event: FallbackProviderEvents) => {
+        if (event.action === 'fallback-provider:request') {
+          retryAttempt = event.retryAttempt;
+        }
+      });
+
+      const block = await mockedProvider.getBlock(42);
+
+      expect(retryAttempt).toBe(0);
+      expect(block.hash).toBe(fixtures.eth_getBlockByNumber.default.hash);
+    });
   });
 });