Simplify worker block dispatcher

packages/node-core/src/blockchain.service.ts

@@ -2,31 +2,44 @@
 // SPDX-License-Identifier: GPL-3.0
 
 import {BaseCustomDataSource, BaseDataSource, IProjectNetworkConfig} from '@subql/types-core';
-import {DatasourceParams, Header, IBlock, ISubqueryProject} from './indexer';
+import {DatasourceParams, Header, IBaseIndexerWorker, IBlock, ISubqueryProject} from './indexer';
 
 // TODO probably need to split this in 2 to have a worker specific subset
 
+export interface ICoreBlockchainService<
+  DS extends BaseDataSource = BaseDataSource,
+  SubQueryProject extends ISubqueryProject<IProjectNetworkConfig, DS> = ISubqueryProject<IProjectNetworkConfig, DS>
+> {
+  /* The semver of the node */
+  packageVersion: string;
+
+  // Project service
+  onProjectChange(project: SubQueryProject): Promise<void> | void;
+  /* Not all networks have a block timestamp, e.g. Shiden */
+  getBlockTimestamp(height: number): Promise<Date | undefined>;
+}
+
 export interface IBlockchainService<
   DS extends BaseDataSource = BaseDataSource,
   CDS extends DS & BaseCustomDataSource = BaseCustomDataSource & DS,
   SubQueryProject extends ISubqueryProject<IProjectNetworkConfig, DS> = ISubqueryProject<IProjectNetworkConfig, DS>,
   SafeAPI = any,
   LightBlock = any,
   FullBlock = any,
-> {
-  /* The semver of the node */
-  packageVersion: string;
-
+  Worker extends IBaseIndexerWorker = IBaseIndexerWorker
+> extends ICoreBlockchainService<DS, SubQueryProject> {
   blockHandlerKind: string;
   // TODO SubqueryProject methods
 
   // Block dispatcher service
   fetchBlocks(blockNums: number[]): Promise<IBlock<LightBlock>[] | IBlock<FullBlock>[]>; // TODO this probably needs to change to get light block type correct
+  /* This is the worker equivalent of fetchBlocks, it provides a context to allow syncing anything between workers */
+  fetchBlockWorker(worker: Worker, blockNum: number, context: {workers: Worker[]}): Promise<void>;
 
   // Project service
-  onProjectChange(project: SubQueryProject): Promise<void> | void;
-  /* Not all networks have a block timestamp, e.g. Shiden */
-  getBlockTimestamp(height: number): Promise<Date | undefined>;
+  // onProjectChange(project: SubQueryProject): Promise<void> | void;
+  // /* Not all networks have a block timestamp, e.g. Shiden */
+  // getBlockTimestamp(height: number): Promise<Date | undefined>;
 
   // Fetch service
   /**

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

@@ -27,6 +27,7 @@ export type ProcessBlockResponse = {
 };
 
 export interface IBlockDispatcher<B> {
+  init(onDynamicDsCreated: (height: number) => void): Promise<void>;
   // now within enqueueBlock should handle getLatestBufferHeight
   enqueueBlocks(heights: (IBlock<B> | number)[], latestBufferHeight: number): void | Promise<void>;
   queueSize: number;

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

@@ -2,14 +2,19 @@
 // SPDX-License-Identifier: GPL-3.0
 
 import {EventEmitter2} from '@nestjs/event-emitter';
+import {IBlockchainService} from '../../blockchain.service';
 import {IProjectUpgradeService, NodeConfig} from '../../configure';
+import {ConnectionPoolStateManager} from '../connectionPoolState.manager';
+import {DynamicDsService} from '../dynamic-ds.service';
+import {InMemoryCacheService} from '../inMemoryCache.service';
 import {PoiSyncService} from '../poi';
 import {StoreService} from '../store.service';
 import {StoreCacheService} from '../storeCache';
 import {IProjectService, ISubqueryProject} from '../types';
+import {UnfinalizedBlocksService} from '../unfinalizedBlocks.service';
 import {WorkerBlockDispatcher} from './worker-block-dispatcher';
 
-class TestWorkerBlockDispatcher extends WorkerBlockDispatcher<any, any, any> {
+class TestWorkerBlockDispatcher extends WorkerBlockDispatcher<any, any, any, any> {
   async fetchBlock(worker: any, height: number): Promise<void> {
     return Promise.resolve();
   }
@@ -19,7 +24,7 @@ class TestWorkerBlockDispatcher extends WorkerBlockDispatcher<any, any, any> {
   }
 }
 describe('WorkerBlockDispatcher', () => {
-  let dispatcher: WorkerBlockDispatcher<any, any, any>;
+  let dispatcher: WorkerBlockDispatcher<any, any, any, any>;
 
   // Mock workers
   const mockWorkers = [
@@ -36,9 +41,15 @@ describe('WorkerBlockDispatcher', () => {
       null as unknown as IProjectUpgradeService,
       null as unknown as StoreService,
       null as unknown as StoreCacheService,
+      null as unknown as InMemoryCacheService,
       null as unknown as PoiSyncService,
+      null as unknown as DynamicDsService,
+      null as unknown as UnfinalizedBlocksService<any>,
+      null as unknown as ConnectionPoolStateManager<any>,
       null as unknown as ISubqueryProject,
-      null as unknown as () => Promise<any>
+      null as unknown as IBlockchainService<any>,
+      '',
+      []
     );
     (dispatcher as any).workers = mockWorkers;
   });

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

@@ -2,14 +2,27 @@
 // SPDX-License-Identifier: GPL-3.0
 
 import assert from 'assert';
-import {OnApplicationShutdown} from '@nestjs/common';
+import {Inject, Injectable, OnApplicationShutdown} from '@nestjs/common';
 import {EventEmitter2} from '@nestjs/event-emitter';
 import {Interval} from '@nestjs/schedule';
+import {BaseDataSource} from '@subql/types-core';
 import {last} from 'lodash';
+import {IApiConnectionSpecific} from '../../api.service';
+import {IBlockchainService} from '../../blockchain.service';
 import {NodeConfig} from '../../configure';
 import {IProjectUpgradeService} from '../../configure/ProjectUpgrade.service';
 import {IndexerEvent} from '../../events';
-import {IBlock, PoiSyncService} from '../../indexer';
+import {
+  ConnectionPoolStateManager,
+  createIndexerWorker,
+  DynamicDsService,
+  IBaseIndexerWorker,
+  IBlock,
+  InMemoryCacheService,
+  PoiSyncService,
+  TerminateableWorker,
+  UnfinalizedBlocksService,
+} from '../../indexer';
 import {getLogger} from '../../logger';
 import {monitorWrite} from '../../process';
 import {AutoQueue, isTaskFlushedError} from '../../utils';
@@ -22,15 +35,6 @@ 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>;
-};
-
 function initAutoQueue<T>(
   workers: number | undefined,
   batchSize: number,
@@ -41,27 +45,39 @@ function initAutoQueue<T>(
   return new AutoQueue(workers * batchSize * 2, 1, timeout, name);
 }
 
-export abstract class WorkerBlockDispatcher<DS, W extends Worker, B>
-  extends BaseBlockDispatcher<AutoQueue<void>, DS, B>
+@Injectable()
+export class WorkerBlockDispatcher<
+    DS extends BaseDataSource = BaseDataSource,
+    Worker extends IBaseIndexerWorker = IBaseIndexerWorker,
+    Block = any,
+    ApiConn extends IApiConnectionSpecific = IApiConnectionSpecific
+  >
+  extends BaseBlockDispatcher<AutoQueue<void>, DS, Block>
   implements OnApplicationShutdown
 {
-  protected workers: W[] = [];
+  protected workers: TerminateableWorker<Worker>[] = [];
   private numWorkers: number;
   private isShutdown = false;
   private currentWorkerIndex = 0;
 
-  protected abstract fetchBlock(worker: W, height: number): Promise<void>;
+  private createWorker: () => Promise<TerminateableWorker<Worker>>;
 
   constructor(
     nodeConfig: NodeConfig,
     eventEmitter: EventEmitter2,
-    projectService: IProjectService<DS>,
-    projectUpgradeService: IProjectUpgradeService,
+    @Inject('IProjectService') projectService: IProjectService<DS>,
+    @Inject('IProjectUpgradeService') projectUpgradeService: IProjectUpgradeService,
     storeService: StoreService,
     storeCacheService: StoreCacheService,
+    cacheService: InMemoryCacheService,
     poiSyncService: PoiSyncService,
-    project: ISubqueryProject,
-    private createIndexerWorker: () => Promise<W>,
+    dynamicDsService: DynamicDsService<DS>,
+    unfinalizedBlocksService: UnfinalizedBlocksService<Block>,
+    connectionPoolState: ConnectionPoolStateManager<ApiConn>,
+    @Inject('ISubqueryProject') project: ISubqueryProject,
+    @Inject('IBlockchainService') private blockchainService: IBlockchainService<DS>,
+    workerPath: string,
+    workerFns: Parameters<typeof createIndexerWorker<Worker, ApiConn, Block, DS>>[1],
     monitorService?: MonitorServiceInterface
   ) {
     super(
@@ -76,13 +92,27 @@ export abstract class WorkerBlockDispatcher<DS, W extends Worker, B>
       poiSyncService,
       monitorService
     );
+
+    this.createWorker = () =>
+      createIndexerWorker<Worker, ApiConn, Block, DS>(
+        workerPath,
+        workerFns,
+        storeService.getStore(),
+        cacheService.getCache(),
+        dynamicDsService,
+        unfinalizedBlocksService,
+        connectionPoolState,
+        project.root,
+        projectService.startHeight,
+        monitorService
+      );
     // initAutoQueue will assert that workers is set. unfortunately we cant do anything before the super call
     // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
     this.numWorkers = nodeConfig.workers!;
   }
 
   async init(onDynamicDsCreated: (height: number) => void): Promise<void> {
-    this.workers = await Promise.all(new Array(this.numWorkers).fill(0).map(() => this.createIndexerWorker()));
+    this.workers = await Promise.all(new Array(this.numWorkers).fill(0).map(() => this.createWorker()));
     return super.init(onDynamicDsCreated);
   }
 
@@ -96,7 +126,8 @@ export abstract class WorkerBlockDispatcher<DS, W extends Worker, B>
       await Promise.all(this.workers.map((w) => w.terminate()));
     }
   }
-  async enqueueBlocks(heights: (IBlock<B> | number)[], latestBufferHeight?: number): Promise<void> {
+
+  async enqueueBlocks(heights: (IBlock<Block> | number)[], latestBufferHeight?: number): Promise<void> {
     assert(
       heights.every((h) => typeof h === 'number'),
       'Worker block dispatcher only supports enqueuing numbers, not blocks.'
@@ -147,7 +178,7 @@ export abstract class WorkerBlockDispatcher<DS, W extends Worker, B>
 
     await worker.waitForWorkerBatchSize(this.minimumHeapLimit);
 
-    const pendingBlock = this.fetchBlock(worker, height);
+    const pendingBlock = this.blockchainService.fetchBlockWorker(worker, height, {workers: this.workers});
 
     const processBlock = async () => {
       try {

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

@@ -8,7 +8,7 @@ import {EventEmitter2} from '@nestjs/event-emitter';
 import {BaseDataSource, IProjectNetworkConfig} from '@subql/types-core';
 import {Sequelize} from '@subql/x-sequelize';
 import {IApi} from '../api.service';
-import {IBlockchainService} from '../blockchain.service';
+import {IBlockchainService, ICoreBlockchainService} from '../blockchain.service';
 import {IProjectUpgradeService, NodeConfig} from '../configure';
 import {IndexerEvent} from '../events';
 import {getLogger} from '../logger';
@@ -35,7 +35,7 @@ class NotInitError extends Error {
 export class ProjectService<
   DS extends BaseDataSource = BaseDataSource,
   API extends IApi = IApi,
-  UnfinalizedBlocksService extends IUnfinalizedBlocksService<any> = IUnfinalizedBlocksService<any>,
+  UnfinalizedBlocksService extends IUnfinalizedBlocksService<any> = IUnfinalizedBlocksService<any>
 > implements IProjectService<DS>
 {
   private _schema?: string;
@@ -55,7 +55,7 @@ export class ProjectService<
     private readonly dynamicDsService: DynamicDsService<DS>,
     private eventEmitter: EventEmitter2,
     @Inject('IUnfinalizedBlocksService') private readonly unfinalizedBlockService: UnfinalizedBlocksService,
-    @Inject('IBlockchainService') private blockchainService: IBlockchainService<DS>
+    @Inject('IBlockchainService') private blockchainService: ICoreBlockchainService<DS>
   ) {
     if (this.nodeConfig.unfinalizedBlocks && this.nodeConfig.allowSchemaMigration) {
       throw new Error('Unfinalized Blocks and Schema Migration cannot be enabled at the same time');

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

@@ -172,6 +172,8 @@ export function createWorkerHost<
   );
 }
 
+export type TerminateableWorker<T extends IBaseIndexerWorker> = T & {terminate: () => Promise<number>};
+
 export async function createIndexerWorker<
   T extends IBaseIndexerWorker,
   ApiConnection extends IApiConnectionSpecific<any, any, any> /*ApiPromiseConnection*/ /*ApiPromiseConnection*/,
@@ -189,7 +191,7 @@ export async function createIndexerWorker<
   startHeight: number,
   monitorService?: MonitorServiceInterface,
   workerData?: any
-): Promise<T & {terminate: () => Promise<number>}> {
+): Promise<TerminateableWorker<T>> {
   const indexerWorker = Worker.create<
     T & {initWorker: (startHeight: number) => Promise<void>},
     DefaultWorkerFunctions<ApiConnection, DS>

packages/node/src/blockchain.service.ts

@@ -7,6 +7,7 @@ import { isCustomDs, isRuntimeDs } from '@subql/common-substrate';
 import {
   DatasourceParams,
   Header,
+  IBaseIndexerWorker,
   IBlock,
   IBlockchainService,
   mainThreadOnly,
@@ -30,6 +31,7 @@ import {
   isFullBlock,
   LightBlockContent,
 } from './indexer/types';
+import { IIndexerWorker } from './indexer/worker/worker';
 import {
   calcInterval,
   getBlockByHeight,
@@ -55,20 +57,21 @@ export class BlockchainService
       SubqueryProject,
       ApiAt,
       LightBlockContent,
-      BlockContent
+      BlockContent,
+      IIndexerWorker
     >
 {
   constructor(
     @Inject('APIService') private apiService: ApiService,
-    @Inject(isMainThread ? RuntimeService : 'Null')
-    private runtimeService: RuntimeService,
+    @Inject('RuntimeService') private runtimeService: RuntimeService,
   ) {}
 
   isCustomDs = isCustomDs;
   isRuntimeDs = isRuntimeDs;
   blockHandlerKind = SubstrateHandlerKind.Block;
   packageVersion = packageVersion;
 
+  @mainThreadOnly()
   async fetchBlocks(
     blockNums: number[],
   ): Promise<IBlock<BlockContent>[] | IBlock<LightBlockContent>[]> {
@@ -84,6 +87,29 @@ export class BlockchainService
     );
   }
 
+  async fetchBlockWorker(
+    worker: IIndexerWorker,
+    height: number,
+    context: { workers: IIndexerWorker[] },
+  ): Promise<void> {
+    // get SpecVersion from main runtime service
+    const { blockSpecVersion, syncedDictionary } =
+      await this.runtimeService.getSpecVersion(height);
+
+    // if main runtime specVersion has been updated, then sync with all workers specVersion map, and lastFinalizedBlock
+    if (syncedDictionary) {
+      context.workers.map((w) =>
+        w.syncRuntimeService(
+          this.runtimeService.specVersionMap,
+          this.runtimeService.latestFinalizedHeight,
+        ),
+      );
+    }
+
+    // const start = new Date();
+    await worker.fetchBlock(height, blockSpecVersion);
+  }
+
   async onProjectChange(project: SubqueryProject): Promise<void> {
     // Only network with chainTypes require to reload
     await this.apiService.updateChainTypes();
@@ -98,8 +124,9 @@ export class BlockchainService
   async getFinalizedHeader(): Promise<Header> {
     const finalizedHash =
       await this.apiService.unsafeApi.rpc.chain.getFinalizedHead();
-    const finalizedHeader =
-      await this.apiService.unsafeApi.rpc.chain.getHeader(finalizedHash);
+    const finalizedHeader = await this.apiService.unsafeApi.rpc.chain.getHeader(
+      finalizedHash,
+    );
     return substrateHeaderToHeader(finalizedHeader);
   }