Implement dynamic block fetching based on block size

packages/node-core/src/indexer/blockDispatcher/base-block-dispatcher.ts

@@ -13,7 +13,6 @@ import {monitorCreateBlockFork, monitorCreateBlockStart, monitorWrite} from '../
 import {IQueue, mainThreadOnly} from '../../utils';
 import {MonitorServiceInterface} from '../monitor.service';
 import {PoiBlock, PoiSyncService} from '../poi';
-import {SmartBatchService} from '../smartBatch.service';
 import {StoreService} from '../store.service';
 import {IStoreModelProvider} from '../storeModelProvider';
 import {IPoi} from '../storeModelProvider/poi';
@@ -32,8 +31,7 @@ export interface IBlockDispatcher<B> {
   queueSize: number;
   freeSize: number;
   latestBufferedHeight: number;
-  smartBatchSize: number;
-  minimumHeapLimit: number;
+  batchSize: number;
 
   // Remove all enqueued blocks, used when a dynamic ds is created
   flushQueue(height: number): void;
@@ -53,8 +51,6 @@ export abstract class BaseBlockDispatcher<Q extends IQueue, DS, B> implements IB
   private _onDynamicDsCreated?: (height: number) => void;
   private _pendingRewindHeader?: Header;
 
-  protected smartBatchService: SmartBatchService;
-
   constructor(
     protected nodeConfig: NodeConfig,
     protected eventEmitter: EventEmitter2,
@@ -66,9 +62,7 @@ export abstract class BaseBlockDispatcher<Q extends IQueue, DS, B> implements IB
     private storeModelProvider: IStoreModelProvider,
     private poiSyncService: PoiSyncService,
     protected monitorService?: MonitorServiceInterface
-  ) {
-    this.smartBatchService = new SmartBatchService(nodeConfig.batchSize);
-  }
+  ) {}
 
   abstract enqueueBlocks(heights: (IBlock<B> | number)[], latestBufferHeight?: number): void | Promise<void>;
 
@@ -87,12 +81,9 @@ export abstract class BaseBlockDispatcher<Q extends IQueue, DS, B> implements IB
     return this.queue.freeSpace;
   }
 
-  get smartBatchSize(): number {
-    return this.smartBatchService.getSafeBatchSize();
-  }
-
-  get minimumHeapLimit(): number {
-    return this.smartBatchService.minimumHeapRequired;
+  get batchSize(): number {
+    // TODO make this smarter
+    return this.nodeConfig.batchSize;
   }
 
   get latestProcessedHeight(): number {

packages/node-core/src/indexer/blockDispatcher/block-dispatcher.ts

@@ -1,7 +1,6 @@
 // Copyright 2020-2024 SubQuery Pte Ltd authors & contributors
 // SPDX-License-Identifier: GPL-3.0
 
-import {getHeapStatistics} from 'v8';
 import {OnApplicationShutdown} from '@nestjs/common';
 import {EventEmitter2} from '@nestjs/event-emitter';
 import {Interval} from '@nestjs/schedule';
@@ -12,11 +11,12 @@ import {getBlockHeight, IBlock, PoiSyncService} from '../../indexer';
 import {getLogger} from '../../logger';
 import {exitWithError, monitorWrite} from '../../process';
 import {profilerWrap} from '../../profiler';
-import {Queue, AutoQueue, delay, memoryLock, waitForBatchSize, isTaskFlushedError} from '../../utils';
+import {Queue, AutoQueue, RampQueue, delay, isTaskFlushedError} from '../../utils';
 import {StoreService} from '../store.service';
 import {IStoreModelProvider} from '../storeModelProvider';
 import {IProjectService, ISubqueryProject} from '../types';
 import {BaseBlockDispatcher, ProcessBlockResponse} from './base-block-dispatcher';
+// import { RampQueue } from '@subql/node-core/utils/rampQueue';
 
 const logger = getLogger('BlockDispatcherService');
 
@@ -37,6 +37,7 @@ export abstract class BlockDispatcher<B, DS>
   private fetching = false;
   private isShutdown = false;
 
+  protected abstract getBlockSize(block: IBlock<B>): number;
   protected abstract indexBlock(block: IBlock<B>): Promise<ProcessBlockResponse>;
 
   constructor(
@@ -62,7 +63,13 @@ export abstract class BlockDispatcher<B, DS>
       poiSyncService
     );
     this.processQueue = new AutoQueue(nodeConfig.batchSize * 3, 1, nodeConfig.timeout, 'Process');
-    this.fetchQueue = new AutoQueue(nodeConfig.batchSize * 3, nodeConfig.batchSize, nodeConfig.timeout, 'Fetch');
+    this.fetchQueue = new RampQueue(
+      this.getBlockSize.bind(this),
+      nodeConfig.batchSize,
+      nodeConfig.batchSize * 3,
+      nodeConfig.timeout,
+      'Fetch'
+    );
     if (this.nodeConfig.profiler) {
       this.fetchBlocksBatches = profilerWrap(fetchBlocksBatches, 'BlockDispatcher', 'fetchBlocksBatches');
     } else {
@@ -96,17 +103,14 @@ export abstract class BlockDispatcher<B, DS>
     this.processQueue.flush();
   }
 
-  private memoryleft(): number {
-    return this.smartBatchService.heapMemoryLimit() - getHeapStatistics().used_heap_size;
-  }
-
   @Interval(10000)
   queueStats(stat: 'size' | 'freeSpace' = 'freeSpace'): void {
     // NOTE: If the free space of the process queue is low it means that processing is the limiting factor. If it is large then fetching blocks is the limitng factor.
     logger.debug(
       `QUEUE INFO ${stat}: Block numbers: ${this.queue[stat]}, fetch: ${this.fetchQueue[stat]}, process: ${this.processQueue[stat]}`
     );
   }
+
   private async fetchBlocksFromQueue(): Promise<void> {
     if (this.fetching || this.isShutdown) return;
 
@@ -128,23 +132,14 @@ export abstract class BlockDispatcher<B, DS>
 
         // Used to compare before and after as a way to check if queue was flushed
         const bufferedHeight = this._latestBufferedHeight;
-
-        if (this.memoryleft() < 0) {
-          //stop fetching until memory is freed
-          await waitForBatchSize(this.minimumHeapLimit);
-        }
-
         void this.fetchQueue
           .put(async () => {
-            if (memoryLock.isLocked()) {
-              await memoryLock.waitForUnlock();
-            }
             if (typeof blockOrNum !== 'number') {
               // Type is of block
               return blockOrNum;
             }
             const [block] = await this.fetchBlocksBatches([blockOrNum]);
-            this.smartBatchService.addToSizeBuffer([block]);
+
             return block;
           })
           .then(

packages/node-core/src/indexer/blockDispatcher/worker-block-dispatcher.ts

@@ -2,32 +2,30 @@
 // SPDX-License-Identifier: GPL-3.0
 
 import assert from 'assert';
-import { OnApplicationShutdown } from '@nestjs/common';
-import { EventEmitter2 } from '@nestjs/event-emitter';
-import { Interval } from '@nestjs/schedule';
-import { last } from 'lodash';
-import { NodeConfig } from '../../configure';
-import { IProjectUpgradeService } from '../../configure/ProjectUpgrade.service';
-import { IndexerEvent } from '../../events';
-import { IBlock, PoiSyncService } from '../../indexer';
-import { getLogger } from '../../logger';
-import { monitorWrite } from '../../process';
-import { AutoQueue, isTaskFlushedError } from '../../utils';
-import { MonitorServiceInterface } from '../monitor.service';
-import { StoreService } from '../store.service';
-import { IStoreModelProvider } from '../storeModelProvider';
-import { ISubqueryProject, IProjectService, Header } from '../types';
-import { isBlockUnavailableError } from '../worker/utils';
-import { BaseBlockDispatcher } from './base-block-dispatcher';
+import {OnApplicationShutdown} from '@nestjs/common';
+import {EventEmitter2} from '@nestjs/event-emitter';
+import {Interval} from '@nestjs/schedule';
+import {last} from 'lodash';
+import {NodeConfig} from '../../configure';
+import {IProjectUpgradeService} from '../../configure/ProjectUpgrade.service';
+import {IndexerEvent} from '../../events';
+import {IBlock, PoiSyncService} from '../../indexer';
+import {getLogger} from '../../logger';
+import {monitorWrite} from '../../process';
+import {AutoQueue, isTaskFlushedError} from '../../utils';
+import {MonitorServiceInterface} from '../monitor.service';
+import {StoreService} from '../store.service';
+import {IStoreModelProvider} from '../storeModelProvider';
+import {ISubqueryProject, IProjectService, Header} from '../types';
+import {isBlockUnavailableError} from '../worker/utils';
+import {BaseBlockDispatcher} from './base-block-dispatcher';
 
 const logger = getLogger('WorkerBlockDispatcherService');
 
 type Worker = {
   processBlock: (height: number) => Promise<any>;
   getStatus: () => Promise<any>;
   getMemoryLeft: () => Promise<number>;
-  getBlocksLoaded: () => Promise<number>;
-  waitForWorkerBatchSize: (heapSizeInBytes: number) => Promise<void>;
   terminate: () => Promise<number>;
 };
 
@@ -43,11 +41,11 @@ function initAutoQueue<T>(
 
 export abstract class WorkerBlockDispatcher<DS, W extends Worker, B>
   extends BaseBlockDispatcher<AutoQueue<void>, DS, B>
-  implements OnApplicationShutdown {
+  implements OnApplicationShutdown
+{
   protected workers: W[] = [];
   private numWorkers: number;
   private isShutdown = false;
-  private currentWorkerIndex = 0;
 
   protected abstract fetchBlock(worker: W, height: number): Promise<Header>;
 
@@ -115,7 +113,7 @@ export abstract class WorkerBlockDispatcher<DS, W extends Worker, B>
       let startIndex = 0;
       while (startIndex < heights.length) {
         const workerIdx = await this.getNextWorkerIndex();
-        const batchSize = Math.min(heights.length - startIndex, await this.maxBatchSize(workerIdx));
+        const batchSize = heights.length - startIndex;
         await Promise.all(
           heights
             .slice(startIndex, startIndex + batchSize)
@@ -135,6 +133,7 @@ export abstract class WorkerBlockDispatcher<DS, W extends Worker, B>
     this.latestBufferedHeight = latestBufferHeight ?? last(heights as number[]) ?? this.latestBufferedHeight;
   }
 
+  // eslint-disable-next-line @typescript-eslint/require-await
   private async enqueueBlock(height: number, workerIdx: number): Promise<void> {
     if (this.isShutdown) return;
     const worker = this.workers[workerIdx];
@@ -143,9 +142,6 @@ export abstract class WorkerBlockDispatcher<DS, W extends Worker, B>
 
     // Used to compare before and after as a way to check if queue was flushed
     const bufferedHeight = this.latestBufferedHeight;
-
-    await worker.waitForWorkerBatchSize(this.minimumHeapLimit);
-
     const pendingBlock = this.fetchBlock(worker, height);
 
     const processBlock = async () => {
@@ -159,7 +155,7 @@ export abstract class WorkerBlockDispatcher<DS, W extends Worker, B>
         await this.preProcessBlock(header);
 
         monitorWrite(`Processing from worker #${workerIdx}`);
-        const { dynamicDsCreated, reindexBlockHeader } = await worker.processBlock(height);
+        const {dynamicDsCreated, reindexBlockHeader} = await worker.processBlock(height);
 
         await this.postProcessBlock(header, {
           dynamicDsCreated,
@@ -211,25 +207,8 @@ export abstract class WorkerBlockDispatcher<DS, W extends Worker, B>
   }
 
   private async getNextWorkerIndex(): Promise<number> {
-    const startIndex = this.currentWorkerIndex;
-    do {
-      this.currentWorkerIndex = (this.currentWorkerIndex + 1) % this.workers.length;
-      const memLeft = await this.workers[this.currentWorkerIndex].getMemoryLeft();
-      if (memLeft >= this.minimumHeapLimit) {
-        return this.currentWorkerIndex;
-      }
-    } while (this.currentWorkerIndex !== startIndex);
-
-    // All workers have been tried and none have enough memory left.
-    // wait for any worker to free the memory before calling getNextWorkerIndex again
-    await Promise.race(this.workers.map((worker) => worker.waitForWorkerBatchSize(this.minimumHeapLimit)));
-
-    return this.getNextWorkerIndex();
-  }
-
-  private async maxBatchSize(workerIdx: number): Promise<number> {
-    const memLeft = await this.workers[workerIdx].getMemoryLeft();
-    if (memLeft < this.minimumHeapLimit) return 0;
-    return this.smartBatchService.safeBatchSizeForRemainingMemory(memLeft);
+    return Promise.all(this.workers.map((worker) => worker.getMemoryLeft())).then((memoryLeftValues) => {
+      return memoryLeftValues.indexOf(Math.max(...memoryLeftValues));
+    });
   }
 }

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

@@ -4,7 +4,6 @@
 import {EventEmitter2} from '@nestjs/event-emitter';
 import {SchedulerRegistry} from '@nestjs/schedule';
 import {BaseDataSource, BaseHandler, BaseMapping, DictionaryQueryEntry} from '@subql/types-core';
-import {range} from 'lodash';
 import {
   BaseUnfinalizedBlocksService,
   BlockDispatcher,
@@ -145,8 +144,7 @@ const getDictionaryService = () =>
 const getBlockDispatcher = () => {
   const inst = {
     latestBufferedHeight: 0,
-    smartBatchSize: 10,
-    minimumHeapLimit: 1000,
+    batchSize: 10,
     freeSize: 10,
     enqueueBlocks: (heights: number[], latestBufferHeight: number) => {
       (inst as any).freeSize = inst.freeSize - heights.length;

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

@@ -10,7 +10,7 @@ import {range} from 'lodash';
 import {NodeConfig} from '../configure';
 import {IndexerEvent} from '../events';
 import {getLogger} from '../logger';
-import {delay, filterBypassBlocks, waitForBatchSize} from '../utils';
+import {delay, filterBypassBlocks} from '../utils';
 import {IBlockDispatcher} from './blockDispatcher';
 import {mergeNumAndBlocksToNums} from './dictionary';
 import {DictionaryService} from './dictionary/dictionary.service';
@@ -28,6 +28,9 @@ export abstract class BaseFetchService<DS extends BaseDataSource, B extends IBlo
   private _latestFinalizedHeight?: number;
   private isShutdown = false;
 
+  #pendingFinalizedBlockHead?: Promise<void>;
+  #pendingBestBlockHead?: Promise<void>;
+
   // If the chain doesn't have a distinction between the 2 it should return the same value for finalized and best
   protected abstract getFinalizedHeader(): Promise<Header>;
   protected abstract getBestHeight(): Promise<number>;
@@ -77,10 +80,24 @@ export abstract class BaseFetchService<DS extends BaseDataSource, B extends IBlo
     this.isShutdown = true;
   }
 
+  // Memoizes the request by not making new ones if one is already in progress
+  private async memoGetFinalizedBlockHead(): Promise<void> {
+    return (this.#pendingFinalizedBlockHead ??= this.getFinalizedBlockHead().finally(
+      () => (this.#pendingFinalizedBlockHead = undefined)
+    ));
+  }
+
+  // Memoizes the request by not making new ones if one is already in progress
+  private async memoGetBestBlockHead(): Promise<void> {
+    return (this.#pendingBestBlockHead ??= this.getBestBlockHead().finally(
+      () => (this.#pendingBestBlockHead = undefined)
+    ));
+  }
+
   async init(startHeight: number): Promise<void> {
     const interval = await this.getChainInterval();
 
-    await Promise.all([this.getFinalizedBlockHead(), this.getBestBlockHead()]);
+    await Promise.all([this.memoGetFinalizedBlockHead(), this.memoGetBestBlockHead()]);
 
     const chainLatestHeight = this.latestHeight();
     if (startHeight > chainLatestHeight) {
@@ -100,11 +117,11 @@ export abstract class BaseFetchService<DS extends BaseDataSource, B extends IBlo
 
     this.schedulerRegistry.addInterval(
       'getFinalizedBlockHead',
-      setInterval(() => void this.getFinalizedBlockHead(), interval)
+      setInterval(() => void this.memoGetFinalizedBlockHead(), interval)
     );
     this.schedulerRegistry.addInterval(
       'getBestBlockHead',
-      setInterval(() => void this.getBestBlockHead(), interval)
+      setInterval(() => void this.memoGetBestBlockHead(), interval)
     );
 
     await this.dictionaryService.initDictionaries();
@@ -186,12 +203,7 @@ export abstract class BaseFetchService<DS extends BaseDataSource, B extends IBlo
     while (!this.isShutdown) {
       startBlockHeight = getStartBlockHeight();
 
-      scaledBatchSize = this.blockDispatcher.smartBatchSize;
-
-      if (scaledBatchSize === 0) {
-        await waitForBatchSize(this.blockDispatcher.minimumHeapLimit);
-        continue;
-      }
+      scaledBatchSize = this.blockDispatcher.batchSize;
 
       const latestHeight = this.latestHeight();
 
@@ -211,7 +223,7 @@ export abstract class BaseFetchService<DS extends BaseDataSource, B extends IBlo
       }
 
       // Update the target height, this happens here to stay in sync with the rest of indexing
-      this.storeModelProvider.metadata.set('targetHeight', latestHeight);
+      await this.storeModelProvider.metadata.set('targetHeight', latestHeight);
 
       // This could be latestBestHeight, dictionary should never include finalized blocks
       // TODO add buffer so dictionary not used when project synced

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

@@ -186,7 +186,7 @@ export abstract class BaseIndexerManager<
 
       for (const handler of handlers) {
         // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-        vm = vm! ?? (await getVM(ds));
+        vm ??= await getVM(ds);
 
         const parsedData = await this.prepareFilteredData(kind, data, ds);
 
@@ -208,7 +208,7 @@ export abstract class BaseIndexerManager<
 
       for (const handler of handlers) {
         // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-        vm = vm! ?? (await getVM(ds));
+        vm ??= await getVM(ds);
         monitorWrite(() => `- Handler: ${handler.handler}, args:${handledStringify(data)}`);
         await this.transformAndExecuteCustomDs(ds, vm, handler, data);
       }