Sync 20221117

package.json

@@ -33,7 +33,7 @@
     "typescript": "^4.4.4"
   },
   "resolutions": {
-    "@polkadot/api": "9.5.2",
+    "@polkadot/api": "9.7.1",
     "@polkadot/util": "10.1.11",
     "@terra-money/terra.js": "^3.0.11",
     "node-fetch": "2.6.7"

packages/node/package.json

@@ -26,7 +26,7 @@
     "@nestjs/schedule": "^1.0.2",
     "@subql/common": "latest",
     "@subql/common-ethereum": "workspace:*",
-    "@subql/node-core": "1.3.3",
+    "@subql/node-core": "1.4.2-0",
     "@subql/types-ethereum": "workspace:*",
     "@subql/utils": "latest",
     "@subql/x-merkle-mountain-range": "2.0.0-0.1.2",

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

@@ -0,0 +1,146 @@
+// Copyright 2020-2021 OnFinality Limited authors & contributors
+// SPDX-License-Identifier: Apache-2.0
+
+import assert from 'assert';
+import { EventEmitter2 } from '@nestjs/event-emitter';
+import { hexToU8a, u8aEq } from '@polkadot/util';
+import { getLogger, IndexerEvent, IQueue, NodeConfig } from '@subql/node-core';
+import { ProjectService } from '../project.service';
+
+const logger = getLogger('BaseBlockDispatcherService');
+
+export type ProcessBlockResponse = {
+  dynamicDsCreated: boolean;
+  operationHash: Uint8Array;
+  reindexBlockHeight: number;
+};
+
+export interface IBlockDispatcher {
+  init(onDynamicDsCreated: (height: number) => Promise<void>): Promise<void>;
+
+  enqueueBlocks(heights: number[]): void;
+
+  queueSize: number;
+  freeSize: number;
+  latestBufferedHeight: number | undefined;
+
+  // Remove all enqueued blocks, used when a dynamic ds is created
+  flushQueue(height: number): void;
+  rewind(height: number): Promise<void>;
+}
+
+const NULL_MERKEL_ROOT = hexToU8a('0x00');
+
+function isNullMerkelRoot(operationHash: Uint8Array): boolean {
+  return u8aEq(operationHash, NULL_MERKEL_ROOT);
+}
+
+export abstract class BaseBlockDispatcher<Q extends IQueue>
+  implements IBlockDispatcher
+{
+  protected _latestBufferedHeight: number;
+  protected _processedBlockCount: number;
+  protected latestProcessedHeight: number;
+  protected currentProcessingHeight: number;
+  protected onDynamicDsCreated: (height: number) => Promise<void>;
+
+  constructor(
+    protected nodeConfig: NodeConfig,
+    protected eventEmitter: EventEmitter2,
+    protected projectService: ProjectService,
+    protected queue: Q,
+  ) {}
+
+  abstract enqueueBlocks(heights: number[]): void;
+  abstract init(
+    onDynamicDsCreated: (height: number) => Promise<void>,
+  ): Promise<void>;
+
+  get queueSize(): number {
+    return this.queue.size;
+  }
+
+  get freeSize(): number {
+    return this.queue.freeSpace;
+  }
+
+  get latestBufferedHeight(): number {
+    return this._latestBufferedHeight;
+  }
+
+  set latestBufferedHeight(height: number) {
+    this.eventEmitter.emit(IndexerEvent.BlocknumberQueueSize, {
+      value: this.queueSize,
+    });
+    this._latestBufferedHeight = height;
+  }
+
+  protected setProcessedBlockCount(processedBlockCount: number): void {
+    this._processedBlockCount = processedBlockCount;
+    this.eventEmitter.emit(IndexerEvent.BlockProcessedCount, {
+      processedBlockCount,
+      timestamp: Date.now(),
+    });
+  }
+
+  //  Compare it with current indexing number, if last corrected is already indexed
+  //  rewind, also flush queued blocks, drop current indexing transaction, set last processed to correct block too
+  //  if rollback is greater than current index flush queue only
+  async rewind(lastCorrectHeight: number): Promise<void> {
+    if (lastCorrectHeight <= this.currentProcessingHeight) {
+      logger.info(
+        `Found last verified block at height ${lastCorrectHeight}, rewinding...`,
+      );
+      await this.projectService.reindex(lastCorrectHeight);
+      this.latestProcessedHeight = lastCorrectHeight;
+      logger.info(`Successful rewind to block ${lastCorrectHeight}!`);
+    }
+    this.flushQueue(lastCorrectHeight);
+    logger.info(`Queued blocks flushed!`); //Also last buffered height reset, next fetching should start after lastCorrectHeight
+  }
+
+  flushQueue(height: number): void {
+    this.latestBufferedHeight = height;
+    this.queue.flush();
+  }
+
+  // Is called directly before a block is processed
+  protected preProcessBlock(height: number): void {
+    this.currentProcessingHeight = height;
+    this.eventEmitter.emit(IndexerEvent.BlockProcessing, {
+      height,
+      timestamp: Date.now(),
+    });
+  }
+
+  // Is called directly after a block is processed
+  protected async postProcessBlock(
+    height: number,
+    processBlockResponse: ProcessBlockResponse,
+  ): Promise<void> {
+    const { dynamicDsCreated, operationHash, reindexBlockHeight } =
+      processBlockResponse;
+    if (reindexBlockHeight !== null && reindexBlockHeight !== undefined) {
+      await this.rewind(reindexBlockHeight);
+      this.latestProcessedHeight = reindexBlockHeight;
+    } else {
+      if (this.nodeConfig.proofOfIndex && !isNullMerkelRoot(operationHash)) {
+        if (!this.projectService.blockOffset) {
+          // Which means during project init, it has not found offset and set value
+          await this.projectService.upsertMetadataBlockOffset(height - 1);
+        }
+        void this.projectService.setBlockOffset(height - 1);
+      }
+      if (dynamicDsCreated) {
+        await this.onDynamicDsCreated(height);
+      }
+      assert(
+        !this.latestProcessedHeight || height > this.latestProcessedHeight,
+        `Block processed out of order. Height: ${height}. Latest: ${this.latestProcessedHeight}`,
+      );
+      // In memory _processedBlockCount increase, db metadata increase BlockCount in indexer.manager
+      this.setProcessedBlockCount(this._processedBlockCount + 1);
+      this.latestProcessedHeight = height;
+    }
+  }
+}

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

@@ -0,0 +1,179 @@
+// Copyright 2020-2021 OnFinality Limited authors & contributors
+// SPDX-License-Identifier: Apache-2.0
+
+import { Injectable, OnApplicationShutdown } from '@nestjs/common';
+import { EventEmitter2 } from '@nestjs/event-emitter';
+import {
+  ApiService,
+  getLogger,
+  NodeConfig,
+  IndexerEvent,
+  delay,
+  profilerWrap,
+  AutoQueue,
+  Queue,
+} from '@subql/node-core';
+import { last } from 'lodash';
+import { IndexerManager } from '../indexer.manager';
+import { ProjectService } from '../project.service';
+import { BaseBlockDispatcher } from './base-block-dispatcher';
+
+const logger = getLogger('BlockDispatcherService');
+
+/**
+ * @description Intended to behave the same as WorkerBlockDispatcherService but doesn't use worker threads or any parallel processing
+ */
+@Injectable()
+export class BlockDispatcherService
+  extends BaseBlockDispatcher<Queue<number>>
+  implements OnApplicationShutdown
+{
+  private processQueue: AutoQueue<void>;
+
+  private fetching = false;
+  private isShutdown = false;
+  private fetchBlocksBatches: ApiService['api']['fetchBlocks'];
+
+  constructor(
+    private apiService: ApiService,
+    nodeConfig: NodeConfig,
+    private indexerManager: IndexerManager,
+    eventEmitter: EventEmitter2,
+    projectService: ProjectService,
+  ) {
+    super(
+      nodeConfig,
+      eventEmitter,
+      projectService,
+      new Queue(nodeConfig.batchSize * 3),
+    );
+    this.processQueue = new AutoQueue(nodeConfig.batchSize * 3);
+
+    const fetchBlocks = this.apiService.api.fetchBlocks.bind(
+      this.apiService.api,
+    );
+    if (this.nodeConfig.profiler) {
+      this.fetchBlocksBatches = profilerWrap(
+        fetchBlocks,
+        'EthereumUtil',
+        'fetchBlocksBatches',
+      );
+    } else {
+      this.fetchBlocksBatches = fetchBlocks;
+    }
+  }
+
+  // eslint-disable-next-line @typescript-eslint/require-await
+  async init(
+    onDynamicDsCreated: (height: number) => Promise<void>,
+  ): Promise<void> {
+    this.onDynamicDsCreated = onDynamicDsCreated;
+    const blockAmount = await this.projectService.getProcessedBlockCount();
+    this.setProcessedBlockCount(blockAmount ?? 0);
+  }
+
+  onApplicationShutdown(): void {
+    this.isShutdown = true;
+    this.processQueue.abort();
+  }
+
+  enqueueBlocks(heights: number[]): void {
+    if (!heights.length) return;
+
+    logger.info(
+      `Enqueing blocks ${heights[0]}...${last(heights)}, total ${
+        heights.length
+      } blocks`,
+    );
+
+    this.queue.putMany(heights);
+    this.latestBufferedHeight = last(heights);
+
+    void this.fetchBlocksFromQueue().catch((e) => {
+      logger.error(e, 'Failed to fetch blocks from queue');
+      if (!this.isShutdown) {
+        process.exit(1);
+      }
+    });
+  }
+
+  flushQueue(height: number): void {
+    super.flushQueue(height);
+    this.processQueue.flush();
+  }
+
+  private async fetchBlocksFromQueue(): Promise<void> {
+    if (this.fetching || this.isShutdown) return;
+    // Process queue is full, no point in fetching more blocks
+    // if (this.processQueue.freeSpace < this.nodeConfig.batchSize) return;
+
+    this.fetching = true;
+
+    try {
+      while (!this.isShutdown) {
+        const blockNums = this.queue.takeMany(
+          Math.min(this.nodeConfig.batchSize, this.processQueue.freeSpace),
+        );
+        // Used to compare before and after as a way to check if queue was flushed
+        const bufferedHeight = this._latestBufferedHeight;
+
+        // Queue is empty
+        if (!blockNums.length) {
+          // The process queue might be full so no block nums were taken, wait and try again
+          if (this.queue.size) {
+            await delay(1);
+            continue;
+          }
+          break;
+        }
+
+        logger.info(
+          `fetch block [${blockNums[0]},${
+            blockNums[blockNums.length - 1]
+          }], total ${blockNums.length} blocks`,
+        );
+
+        const blocks = await this.fetchBlocksBatches(blockNums);
+
+        if (bufferedHeight > this._latestBufferedHeight) {
+          logger.debug(`Queue was reset for new DS, discarding fetched blocks`);
+          continue;
+        }
+        const blockTasks = blocks.map((block) => async () => {
+          const height = block.block.block.header.number.toNumber();
+          try {
+            this.preProcessBlock(height);
+
+            const processBlockResponse = await this.indexerManager.indexBlock(
+              block,
+            );
+
+            await this.postProcessBlock(height, processBlockResponse);
+          } catch (e) {
+            if (this.isShutdown) {
+              return;
+            }
+            logger.error(
+              e,
+              `failed to index block at height ${height} ${
+                e.handler ? `${e.handler}(${e.stack ?? ''})` : ''
+              }`,
+            );
+            throw e;
+          }
+        });
+
+        // There can be enough of a delay after fetching blocks that shutdown could now be true
+        if (this.isShutdown) break;
+
+        this.processQueue.putMany(blockTasks);
+
+        this.eventEmitter.emit(IndexerEvent.BlockQueueSize, {
+          value: this.processQueue.size,
+        });
+      }
+    } finally {
+      this.fetching = false;
+    }
+  }
+}

packages/node/src/indexer/blockDispatcher/index.ts

@@ -0,0 +1,12 @@
+// Copyright 2020-2021 OnFinality Limited authors & contributors
+// SPDX-License-Identifier: Apache-2.0
+
+import { IBlockDispatcher } from './base-block-dispatcher';
+import { BlockDispatcherService } from './block-dispatcher.service';
+import { WorkerBlockDispatcherService } from './worker-block-dispatcher.service';
+
+export {
+  IBlockDispatcher,
+  BlockDispatcherService,
+  WorkerBlockDispatcherService,
+};