Dictionary v2

packages/node-core/CHANGELOG.md

@@ -8,6 +8,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Removed
 - `scale-batch-size` flag as it had no use (#2275)
 
+### Added
+- Add service to support for dictionary v2 (#2257)
+
 ## [7.3.0] - 2024-02-23
 ### Added
 - Schema Migration support for Enums, Relations, Subscription (#2251)

packages/node-core/src/configure/NodeConfig.ts

@@ -23,7 +23,7 @@ export interface IConfig {
   readonly preferRange: boolean;
   readonly networkEndpoint?: string[];
   readonly primaryNetworkEndpoint?: string;
-  readonly networkDictionary?: string;
+  readonly networkDictionary?: string[];
   readonly dictionaryResolver?: string | false;
   readonly dictionaryRegistry: string;
   readonly outputFmt?: 'json';
@@ -143,8 +143,12 @@ export class NodeConfig<C extends IConfig = IConfig> implements IConfig {
     return this._config.primaryNetworkEndpoint;
   }
 
-  get networkDictionary(): string | undefined {
-    return this._config.networkDictionary;
+  get networkDictionaries(): string[] | undefined | false {
+    return typeof this._config.networkDictionary === 'string'
+      ? this._config.networkDictionary === 'false'
+        ? false
+        : [this._config.networkDictionary]
+      : this._config.networkDictionary;
   }
 
   get allowSchemaMigration(): boolean {

packages/node-core/src/configure/configure.module.ts

@@ -137,7 +137,7 @@ export async function registerApp<P extends ISubqueryProject>(
     omitBy(
       {
         endpoint: config.networkEndpoints,
-        dictionary: config.networkDictionary,
+        dictionary: config.networkDictionaries,
       },
       isNil
     )
@@ -160,7 +160,7 @@ export async function registerApp<P extends ISubqueryProject>(
         // Apply the network endpoint and dictionary from the source project to the parent projects if they are not defined in the config
         {
           endpoint: config.networkEndpoints ?? project.network.endpoint,
-          dictionary: config.networkDictionary ?? project.network.dictionary,
+          dictionary: config.networkDictionaries ?? project.network.dictionary,
         },
         isNil
       )

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

@@ -4,6 +4,7 @@
 import assert from 'assert';
 
 import {EventEmitter2} from '@nestjs/event-emitter';
+import {IBlock} from '@subql/types-core';
 import {hexToU8a, u8aEq} from '@subql/utils';
 import {NodeConfig, IProjectUpgradeService} from '../../configure';
 import {IndexerEvent, PoiEvent} from '../../events';
@@ -25,9 +26,9 @@ export type ProcessBlockResponse = {
   reindexBlockHeight: number | null;
 };
 
-export interface IBlockDispatcher {
-  enqueueBlocks(heights: number[], latestBufferHeight?: number): void | Promise<void>;
-
+export interface IBlockDispatcher<B> {
+  // now within enqueueBlock should handle getLatestBufferHeight
+  enqueueBlocks(heights: (IBlock<B> | number)[], latestBufferHeight: number): void | Promise<void>;
   queueSize: number;
   freeSize: number;
   latestBufferedHeight: number;
@@ -44,7 +45,7 @@ function isNullMerkelRoot(operationHash: Uint8Array): boolean {
   return u8aEq(operationHash, NULL_MERKEL_ROOT);
 }
 
-export abstract class BaseBlockDispatcher<Q extends IQueue, DS> implements IBlockDispatcher {
+export abstract class BaseBlockDispatcher<Q extends IQueue, DS, B> implements IBlockDispatcher<B> {
   protected _latestBufferedHeight = 0;
   protected _processedBlockCount = 0;
   protected _latestProcessedHeight = 0;
@@ -65,7 +66,7 @@ export abstract class BaseBlockDispatcher<Q extends IQueue, DS> implements IBloc
     protected dynamicDsService: DynamicDsService<any>
   ) {}
 
-  abstract enqueueBlocks(heights: number[], latestBufferHeight?: number): void | Promise<void>;
+  abstract enqueueBlocks(heights: (IBlock<B> | number)[], latestBufferHeight?: number): void | Promise<void>;
 
   async init(onDynamicDsCreated: (height: number) => Promise<void>): Promise<void> {
     this._onDynamicDsCreated = onDynamicDsCreated;

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

@@ -5,16 +5,15 @@ import {getHeapStatistics} from 'v8';
 import {OnApplicationShutdown} from '@nestjs/common';
 import {EventEmitter2} from '@nestjs/event-emitter';
 import {Interval} from '@nestjs/schedule';
-import {last} from 'lodash';
+import {IBlock} from '@subql/types-core';
 import {NodeConfig} from '../../configure';
 import {IProjectUpgradeService} from '../../configure/ProjectUpgrade.service';
 import {IndexerEvent} from '../../events';
-import {PoiSyncService} from '../../indexer';
+import {getBlockHeight, PoiSyncService} from '../../indexer';
 import {getLogger} from '../../logger';
 import {profilerWrap} from '../../profiler';
 import {Queue, AutoQueue, delay, memoryLock, waitForBatchSize, isTaskFlushedError} from '../../utils';
 import {DynamicDsService} from '../dynamic-ds.service';
-import {PoiService} from '../poi/poi.service';
 import {SmartBatchService} from '../smartBatch.service';
 import {StoreService} from '../store.service';
 import {StoreCacheService} from '../storeCache';
@@ -23,25 +22,24 @@ import {BaseBlockDispatcher, ProcessBlockResponse} from './base-block-dispatcher
 
 const logger = getLogger('BlockDispatcherService');
 
-type BatchBlockFetcher<B> = (heights: number[]) => Promise<B[]>;
+type BatchBlockFetcher<B> = (heights: number[]) => Promise<IBlock<B>[]>;
 
 /**
  * @description Intended to behave the same as WorkerBlockDispatcherService but doesn't use worker threads or any parallel processing
  */
 export abstract class BlockDispatcher<B, DS>
-  extends BaseBlockDispatcher<Queue<number>, DS>
+  extends BaseBlockDispatcher<Queue<IBlock<B> | number>, DS, B>
   implements OnApplicationShutdown
 {
-  private fetchQueue: AutoQueue<B>;
+  private fetchQueue: AutoQueue<IBlock<B>>;
   private processQueue: AutoQueue<void>;
 
   private fetchBlocksBatches: BatchBlockFetcher<B>;
 
   private fetching = false;
   private isShutdown = false;
 
-  protected abstract indexBlock(block: B): Promise<ProcessBlockResponse>;
-  protected abstract getBlockHeight(block: B): number;
+  protected abstract indexBlock(block: IBlock<B>): Promise<ProcessBlockResponse>;
 
   constructor(
     nodeConfig: NodeConfig,
@@ -71,7 +69,6 @@ export abstract class BlockDispatcher<B, DS>
     );
     this.processQueue = new AutoQueue(nodeConfig.batchSize * 3, 1, nodeConfig.timeout, 'Process');
     this.fetchQueue = new AutoQueue(nodeConfig.batchSize * 3, nodeConfig.batchSize, nodeConfig.timeout, 'Fetch');
-
     if (this.nodeConfig.profiler) {
       this.fetchBlocksBatches = profilerWrap(fetchBlocksBatches, 'BlockDispatcher', 'fetchBlocksBatches');
     } else {
@@ -84,19 +81,18 @@ export abstract class BlockDispatcher<B, DS>
     this.processQueue.abort();
   }
 
-  enqueueBlocks(heights: number[], latestBufferHeight?: number): void {
+  enqueueBlocks(heights: (IBlock<B> | number)[], latestBufferHeight: number): void {
     // In the case where factors of batchSize is equal to bypassBlock or when heights is []
     // to ensure block is bypassed, we set the latestBufferHeight to the heights
     // make sure lastProcessedHeight in metadata is updated
-    if (!!latestBufferHeight && !heights.length) {
+    if (!heights.length) {
       heights = [latestBufferHeight];
     }
-    logger.info(`Enqueueing blocks ${heights[0]}...${last(heights)}, total ${heights.length} blocks`);
-
-    // Those blocks will still be filtered in the handler
+    const startBlockHeight = getBlockHeight(heights[0]);
+    const endBlockHeight = getBlockHeight(heights[heights.length - 1]);
+    logger.info(`Enqueueing blocks ${startBlockHeight}...${endBlockHeight}, total ${heights.length} blocks`);
     this.queue.putMany(heights);
-
-    this.latestBufferedHeight = latestBufferHeight ?? last(heights) ?? this.latestBufferedHeight;
+    this.latestBufferedHeight = latestBufferHeight;
     void this.fetchBlocksFromQueue();
   }
 
@@ -117,7 +113,6 @@ export abstract class BlockDispatcher<B, DS>
       `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;
 
@@ -131,11 +126,9 @@ export abstract class BlockDispatcher<B, DS>
           await delay(1);
           continue;
         }
-
-        const blockNum = this.queue.take();
-
+        const blockOrNum = this.queue.take();
         // This shouldn't happen but if it does it whould get caught above
-        if (!blockNum) {
+        if (!blockOrNum) {
           continue;
         }
 
@@ -152,20 +145,23 @@ export abstract class BlockDispatcher<B, DS>
             if (memoryLock.isLocked()) {
               await memoryLock.waitForUnlock();
             }
-            const [block] = await this.fetchBlocksBatches([blockNum]);
-
+            if (typeof blockOrNum !== 'number') {
+              // Type is of block
+              return blockOrNum;
+            }
+            const [block] = await this.fetchBlocksBatches([blockOrNum]);
             this.smartBatchService.addToSizeBuffer([block]);
             return block;
           })
           .then(
             (block) => {
-              const height = this.getBlockHeight(block);
+              const {height} = block.getHeader();
 
               return this.processQueue.put(async () => {
                 // Check if the queues have been flushed between queue.takeMany and fetchBlocksBatches resolving
                 // Peeking the queue is because the latestBufferedHeight could have regrown since fetching block
                 const peeked = this.queue.peek();
-                if (bufferedHeight > this._latestBufferedHeight || (peeked && peeked < blockNum)) {
+                if (bufferedHeight > this._latestBufferedHeight || (peeked && peeked < height)) {
                   logger.info(`Queue was reset for new DS, discarding fetched blocks`);
                   return;
                 }
@@ -174,7 +170,6 @@ export abstract class BlockDispatcher<B, DS>
                   await this.preProcessBlock(height);
                   // Inject runtimeVersion here to enhance api.at preparation
                   const processBlockResponse = await this.indexBlock(block);
-
                   await this.postProcessBlock(height, processBlockResponse);
 
                   //set block to null for garbage collection
@@ -197,7 +192,7 @@ export abstract class BlockDispatcher<B, DS>
                 // Do nothing, fetching the block was flushed, this could be caused by forked blocks or dynamic datasources
                 return;
               }
-              logger.error(e, `Failed to fetch block ${blockNum}.`);
+              logger.error(e, `Failed to fetch block ${getBlockHeight(blockOrNum)}.`);
               throw e;
             }
           )

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

@@ -11,13 +11,13 @@ import {StoreCacheService} from '../storeCache';
 import {IProjectService, ISubqueryProject} from '../types';
 import {WorkerBlockDispatcher} from './worker-block-dispatcher';
 
-class TestWorkerBlockDispatcher extends WorkerBlockDispatcher<any, any> {
+class TestWorkerBlockDispatcher extends WorkerBlockDispatcher<any, any, any> {
   async fetchBlock(worker: any, height: number): Promise<void> {
     return Promise.resolve();
   }
 }
 describe('WorkerBlockDispatcher', () => {
-  let dispatcher: WorkerBlockDispatcher<any, any>;
+  let dispatcher: WorkerBlockDispatcher<any, any, any>;
 
   // Mock workers
   const mockWorkers = [

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

@@ -5,6 +5,7 @@ import assert from 'assert';
 import {OnApplicationShutdown} from '@nestjs/common';
 import {EventEmitter2} from '@nestjs/event-emitter';
 import {Interval} from '@nestjs/schedule';
+import {IBlock} from '@subql/types-core';
 import {last} from 'lodash';
 import {NodeConfig} from '../../configure';
 import {IProjectUpgradeService} from '../../configure/ProjectUpgrade.service';
@@ -13,7 +14,6 @@ import {PoiSyncService} from '../../indexer';
 import {getLogger} from '../../logger';
 import {AutoQueue, isTaskFlushedError} from '../../utils';
 import {DynamicDsService} from '../dynamic-ds.service';
-import {PoiService} from '../poi/poi.service';
 import {SmartBatchService} from '../smartBatch.service';
 import {StoreService} from '../store.service';
 import {StoreCacheService} from '../storeCache';
@@ -42,8 +42,8 @@ function initAutoQueue<T>(
   return new AutoQueue(workers * batchSize * 2, 1, timeout, name);
 }
 
-export abstract class WorkerBlockDispatcher<DS, W extends Worker>
-  extends BaseBlockDispatcher<AutoQueue<void>, DS>
+export abstract class WorkerBlockDispatcher<DS, W extends Worker, B>
+  extends BaseBlockDispatcher<AutoQueue<void>, DS, B>
   implements OnApplicationShutdown
 {
   protected workers: W[] = [];
@@ -100,8 +100,15 @@ export abstract class WorkerBlockDispatcher<DS, W extends Worker>
       await Promise.all(this.workers.map((w) => w.terminate()));
     }
   }
+  async enqueueBlocks(heights: (IBlock<B> | number)[], latestBufferHeight?: number): Promise<void> {
+    assert(
+      heights.every((h) => typeof h === 'number'),
+      'Worker block dispatcher only supports enqueuing numbers, not blocks.'
+    );
+    await this._enqueueBlocks(heights as number[], latestBufferHeight);
+  }
 
-  async enqueueBlocks(heights: number[], latestBufferHeight?: number): Promise<void> {
+  private async _enqueueBlocks(heights: number[], latestBufferHeight?: number): Promise<void> {
     // In the case where factors of batchSize is equal to bypassBlock or when heights is []
     // to ensure block is bypassed, we set the latestBufferHeight to the heights
     // make sure lastProcessedHeight in metadata is updated