Fix/multi endpint delay

packages/node-core/CHANGELOG.md

@@ -6,6 +6,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+- When configuring multiple endpoints, poor network conditions may lead to block crawling delays. (#2572)
+
 ## [14.1.7] - 2024-10-30
 ### Changed
 - Bump `@subql/common` dependency

packages/node-core/src/indexer/connectionPool.service.spec.ts

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: GPL-3.0
 
 import {delay} from '@subql/common';
-import {ApiErrorType, ConnectionPoolStateManager, IApiConnectionSpecific, NodeConfig} from '..';
+import {ApiErrorType, ConnectionPoolStateManager, getLogger, IApiConnectionSpecific, NodeConfig} from '..';
 import {ConnectionPoolService} from './connectionPool.service';
 
 async function waitFor(conditionFn: () => boolean, timeout = 50000, interval = 100): Promise<void> {

packages/node-core/src/indexer/connectionPool.service.ts

@@ -102,13 +102,6 @@ export class ConnectionPoolService<T extends IApiConnectionSpecific<any, any, an
         if (prop === 'fetchBlocks') {
           return async (heights: number[], ...args: any): Promise<any> => {
             try {
-              // Check if the endpoint is rate-limited
-              if (await this.poolStateManager.getFieldValue(endpoint, 'rateLimited')) {
-                logger.info('throtling on ratelimited endpoint');
-                const backoffDelay = await this.poolStateManager.getFieldValue(endpoint, 'backoffDelay');
-                await delay(backoffDelay / 1000);
-              }
-
               const start = Date.now();
               const result = await target.fetchBlocks(heights, ...args);
               const end = Date.now();

packages/node-core/src/indexer/connectionPoolState.manager.spec.ts

@@ -1,7 +1,8 @@
 // Copyright 2020-2024 SubQuery Pte Ltd authors & contributors
 // SPDX-License-Identifier: GPL-3.0
 
-import {ConnectionPoolStateManager} from './connectionPoolState.manager';
+import {ApiErrorType} from '../api.connection.error';
+import {ConnectionPoolItem, ConnectionPoolStateManager} from './connectionPoolState.manager';
 
 describe('ConnectionPoolStateManager', function () {
   let connectionPoolStateManager: ConnectionPoolStateManager<any>;
@@ -12,62 +13,54 @@ describe('ConnectionPoolStateManager', function () {
     connectionPoolStateManager = new ConnectionPoolStateManager();
   });
 
+  afterEach(async function () {
+    await connectionPoolStateManager.onApplicationShutdown();
+  });
+
   it('chooses primary endpoint first', async function () {
-    (connectionPoolStateManager as any).pool[EXAMPLE_ENDPOINT1] = {
-      primary: true,
-      performanceScore: 100,
-      failureCount: 0,
-      endpoint: '',
-      backoffDelay: 0,
-      rateLimited: false,
-      failed: false,
-      connected: true,
-      lastRequestTime: 0,
-    };
-
-    (connectionPoolStateManager as any).pool[EXAMPLE_ENDPOINT2] = {
-      primary: false,
-      performanceScore: 100,
-      failureCount: 0,
-      endpoint: '',
-      backoffDelay: 0,
-      rateLimited: false,
-      failed: false,
-      connected: true,
-      lastRequestTime: 0,
-    };
+    await connectionPoolStateManager.addToConnections(EXAMPLE_ENDPOINT1, true);
+    await connectionPoolStateManager.addToConnections(EXAMPLE_ENDPOINT2, false);
 
     expect(await connectionPoolStateManager.getNextConnectedEndpoint()).toEqual(EXAMPLE_ENDPOINT1);
   });
 
   it('does not choose primary endpoint if failed', async function () {
-    (connectionPoolStateManager as any).pool[EXAMPLE_ENDPOINT1] = {
-      primary: true,
-      performanceScore: 100,
-      failureCount: 0,
-      endpoint: '',
-      backoffDelay: 0,
-      rateLimited: false,
-      failed: false,
-      connected: false,
-      lastRequestTime: 0,
-    };
-
-    (connectionPoolStateManager as any).pool[EXAMPLE_ENDPOINT2] = {
-      primary: false,
-      performanceScore: 100,
-      failureCount: 0,
-      endpoint: '',
-      backoffDelay: 0,
-      rateLimited: false,
-      failed: false,
-      connected: true,
-      lastRequestTime: 0,
-    };
+    await connectionPoolStateManager.addToConnections(EXAMPLE_ENDPOINT1, true);
+    await connectionPoolStateManager.addToConnections(EXAMPLE_ENDPOINT2, false);
+
+    await connectionPoolStateManager.handleApiError(EXAMPLE_ENDPOINT1, ApiErrorType.Default);
 
     expect(await connectionPoolStateManager.getNextConnectedEndpoint()).toEqual(EXAMPLE_ENDPOINT2);
   });
 
+  it('All endpoints backoff; select a rateLimited endpoint. reason: ApiErrorType.Timeout', async function () {
+    await connectionPoolStateManager.addToConnections(EXAMPLE_ENDPOINT1, false);
+    await connectionPoolStateManager.addToConnections(EXAMPLE_ENDPOINT2, false);
+
+    await connectionPoolStateManager.handleApiError(EXAMPLE_ENDPOINT1, ApiErrorType.Default);
+    await connectionPoolStateManager.handleApiError(EXAMPLE_ENDPOINT2, ApiErrorType.Timeout);
+
+    const nextEndpoint = await connectionPoolStateManager.getNextConnectedEndpoint();
+    const endpointInfo = (connectionPoolStateManager as any).pool[EXAMPLE_ENDPOINT2] as ConnectionPoolItem<any>;
+    expect(nextEndpoint).toEqual(EXAMPLE_ENDPOINT2);
+    expect(endpointInfo.rateLimited).toBe(true);
+    expect((connectionPoolStateManager as any).pool[EXAMPLE_ENDPOINT2].rateLimitDelay).toBe(20 * 1000);
+  });
+
+  it('All endpoints backoff; select a rateLimited endpoint. reason: ApiErrorType.RateLimit', async function () {
+    await connectionPoolStateManager.addToConnections(EXAMPLE_ENDPOINT1, false);
+    await connectionPoolStateManager.addToConnections(EXAMPLE_ENDPOINT2, false);
+
+    await connectionPoolStateManager.handleApiError(EXAMPLE_ENDPOINT1, ApiErrorType.Default);
+    await connectionPoolStateManager.handleApiError(EXAMPLE_ENDPOINT2, ApiErrorType.RateLimit);
+
+    const nextEndpoint = await connectionPoolStateManager.getNextConnectedEndpoint();
+    const endpointInfo = (connectionPoolStateManager as any).pool[EXAMPLE_ENDPOINT2] as ConnectionPoolItem<any>;
+    expect(nextEndpoint).toEqual(EXAMPLE_ENDPOINT2);
+    expect(endpointInfo.rateLimited).toBe(true);
+    expect((connectionPoolStateManager as any).pool[EXAMPLE_ENDPOINT2].rateLimitDelay).toBe(20 * 1000);
+  });
+
   it('can calculate performance score for response time of zero', function () {
     const score = (connectionPoolStateManager as any).calculatePerformanceScore(0, 0);
     expect(score).not.toBeNaN();
@@ -78,4 +71,30 @@ describe('ConnectionPoolStateManager', function () {
     const score2 = (connectionPoolStateManager as any).calculatePerformanceScore(2, 0);
     expect(score1).toBeGreaterThan(score2);
   });
+
+  it('backoff delay rule', function () {
+    const delay1 = (connectionPoolStateManager as any).calculateNextDelay({failureCount: 0});
+    expect(delay1).toBe(0);
+
+    const delay2 = (connectionPoolStateManager as any).calculateNextDelay({failureCount: 1});
+    expect(delay2).toBe(10000);
+
+    const delay3 = (connectionPoolStateManager as any).calculateNextDelay({failureCount: 2});
+    expect(delay3).toBe(20000);
+
+    const delay4 = (connectionPoolStateManager as any).calculateNextDelay({failureCount: 3});
+    expect(delay4).toBe(40000);
+
+    const delay5 = (connectionPoolStateManager as any).calculateNextDelay({failureCount: 4});
+    expect(delay5).toBe(80000);
+
+    const delay6 = (connectionPoolStateManager as any).calculateNextDelay({failureCount: 5});
+    expect(delay6).toBe(160000);
+
+    const delay7 = (connectionPoolStateManager as any).calculateNextDelay({failureCount: 6});
+    expect(delay7).toBe(320000);
+
+    const delayMax = (connectionPoolStateManager as any).calculateNextDelay({failureCount: 7});
+    expect(delayMax).toBe(320000);
+  });
 });

packages/node-core/src/indexer/connectionPoolState.manager.ts

@@ -10,10 +10,10 @@ import {getLogger} from '../logger';
 import {exitWithError} from '../process';
 import {errorTypeToScoreAdjustment} from './connectionPool.service';
 
-const RETRY_DELAY = 60 * 1000;
 const MAX_FAILURES = 5;
 const RESPONSE_TIME_WEIGHT = 0.7;
 const FAILURE_WEIGHT = 0.3;
+const RATE_LIMIT_DELAY = 20 * 1000;
 
 export interface ConnectionPoolItem<T> {
   endpoint: string;
@@ -191,7 +191,7 @@ export class ConnectionPoolStateManager<T extends IApiConnectionSpecific<any, an
   }
 
   //eslint-disable-next-line @typescript-eslint/require-await
-  async setTimeout(endpoint: string, delay: number): Promise<void> {
+  async setRecoverTimeout(endpoint: string, delay: number): Promise<void> {
     // Make sure there is no existing timeout
     await this.clearTimeout(endpoint);
 
@@ -247,35 +247,49 @@ export class ConnectionPoolStateManager<T extends IApiConnectionSpecific<any, an
     switch (errorType) {
       case ApiErrorType.Connection: {
         if (this.pool[endpoint].connected) {
-          //handleApiDisconnects was already called if this is false
-          //this.handleApiDisconnects(endpoint);
+          // The connected status does not provide service. handleApiDisconnects() will be called to handle this.
           this.pool[endpoint].connected = false;
         }
         return;
       }
+      case ApiErrorType.Timeout:
+      case ApiErrorType.RateLimit: {
+        // The “rateLimited” status will be selected when no endpoints are available, so we should avoid setting a large delay.
+        this.pool[endpoint].rateLimited = true;
+        break;
+      }
       case ApiErrorType.Default: {
-        const nextDelay = RETRY_DELAY * Math.pow(2, this.pool[endpoint].failureCount - 1); // Exponential backoff using failure count // Start with RETRY_DELAY and double on each failure
-        this.pool[endpoint].backoffDelay = nextDelay;
-
-        if (ApiErrorType.Timeout || ApiErrorType.RateLimit) {
-          this.pool[endpoint].rateLimited = true;
-        } else {
-          this.pool[endpoint].failed = true;
-        }
-
-        await this.setTimeout(endpoint, nextDelay);
-
-        logger.warn(
-          `Endpoint ${this.pool[endpoint].endpoint} experienced an error (${errorType}). Suspending for ${
-            nextDelay / 1000
-          }s.`
-        );
-        return;
+        // The “failed” status does not provide service.
+        this.pool[endpoint].failed = true;
+        break;
       }
       default: {
         throw new Error(`Unknown error type ${errorType}`);
       }
     }
+
+    const nextDelay = this.calculateNextDelay(this.pool[endpoint]);
+    this.pool[endpoint].backoffDelay = nextDelay;
+    await this.setRecoverTimeout(endpoint, nextDelay);
+
+    logger.warn(
+      `Endpoint ${this.pool[endpoint].endpoint} experienced an error (${errorType}). Suspending for ${
+        nextDelay / 1000
+      }s.`
+    );
+  }
+
+  private calculateNextDelay(poolItem: ConnectionPoolItem<T>): number {
+    const delayRule = [10, 20, 40, 80, 160, 320];
+    let delayTime = 0;
+    if (poolItem.failureCount < 1) {
+      delayTime = 0;
+    } else if (poolItem.failureCount >= 1 && poolItem.failureCount <= delayRule.length) {
+      delayTime = delayRule[poolItem.failureCount - 1];
+    } else {
+      delayTime = delayRule[delayRule.length - 1];
+    }
+    return delayTime * 1000;
   }
 
   private calculatePerformanceScore(responseTime: number, failureCount: number): number {