Implement different fork detection for probabalistic finalization chains

packages/node/src/configure/NodeConfig.ts

@@ -6,6 +6,7 @@ import { IConfig, NodeConfig } from '@subql/node-core';
 export interface IEthereumConfig extends IConfig {
   skipTransactions: boolean;
   blockConfirmations: number;
+  blockForkReindex: number;
 }
 
 export class EthereumNodeConfig extends NodeConfig<IEthereumConfig> {
@@ -32,4 +33,8 @@ export class EthereumNodeConfig extends NodeConfig<IEthereumConfig> {
   get blockConfirmations(): number {
     return this._config.blockConfirmations;
   }
+
+  get blockForkReindex(): number {
+    return this._config.blockForkReindex;
+  }
 }

packages/node/src/ethereum/api.ethereum.ts

@@ -1,7 +1,6 @@
 // Copyright 2020-2023 SubQuery Pte Ltd authors & contributors
 // SPDX-License-Identifier: GPL-3.0
 
-import assert from 'assert';
 import fs from 'fs';
 import http from 'http';
 import https from 'https';

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

@@ -44,7 +44,6 @@ export class BlockDispatcherService
     smartBatchService: SmartBatchService,
     storeService: StoreService,
     storeCacheService: StoreCacheService,
-    poiService: PoiService,
     poiSyncService: PoiSyncService,
     @Inject('ISubqueryProject') project: SubqueryProject,
     dynamicDsService: DynamicDsService,
@@ -57,7 +56,6 @@ export class BlockDispatcherService
       smartBatchService,
       storeService,
       storeCacheService,
-      poiService,
       poiSyncService,
       project,
       dynamicDsService,

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

@@ -48,7 +48,6 @@ export class WorkerBlockDispatcherService
     cacheService: InMemoryCacheService,
     storeService: StoreService,
     storeCacheService: StoreCacheService,
-    poiService: PoiService,
     poiSyncService: PoiSyncService,
     @Inject('ISubqueryProject') project: SubqueryProject,
     dynamicDsService: DynamicDsService,
@@ -63,7 +62,6 @@ export class WorkerBlockDispatcherService
       smartBatchService,
       storeService,
       storeCacheService,
-      poiService,
       poiSyncService,
       project,
       dynamicDsService,

packages/node/src/indexer/fetch.module.ts

@@ -93,7 +93,6 @@ import { UnfinalizedBlocksService } from './unfinalizedBlocks.service';
         cacheService: InMemoryCacheService,
         storeService: StoreService,
         storeCacheService: StoreCacheService,
-        poiService: PoiService,
         poiSyncService: PoiSyncService,
         project: SubqueryProject,
         dynamicDsService: DynamicDsService,
@@ -110,7 +109,6 @@ import { UnfinalizedBlocksService } from './unfinalizedBlocks.service';
               cacheService,
               storeService,
               storeCacheService,
-              poiService,
               poiSyncService,
               project,
               dynamicDsService,
@@ -127,7 +125,6 @@ import { UnfinalizedBlocksService } from './unfinalizedBlocks.service';
               smartBatchService,
               storeService,
               storeCacheService,
-              poiService,
               poiSyncService,
               project,
               dynamicDsService,
@@ -143,7 +140,6 @@ import { UnfinalizedBlocksService } from './unfinalizedBlocks.service';
         InMemoryCacheService,
         StoreService,
         StoreCacheService,
-        PoiService,
         PoiSyncService,
         'ISubqueryProject',
         DynamicDsService,

packages/node/src/indexer/unfinalizedBlocks.service.ts

@@ -10,9 +10,17 @@ import {
   mainThreadOnly,
   NodeConfig,
   StoreCacheService,
+  getLogger,
+  profiler,
+  POI_NOT_ENABLED_ERROR_MESSAGE,
 } from '@subql/node-core';
+import { EthereumBlock } from '@subql/types-ethereum';
+import { last } from 'lodash';
+import { EthereumNodeConfig } from '../configure/NodeConfig';
 import { BlockContent } from './types';
 
+const logger = getLogger('UnfinalizedBlocksService');
+
 export function blockToHeader(block: BlockContent | Block): Header {
   return {
     blockHeight: block.number,
@@ -22,13 +30,119 @@ export function blockToHeader(block: BlockContent | Block): Header {
 }
 
 @Injectable()
-export class UnfinalizedBlocksService extends BaseUnfinalizedBlocksService<BlockContent> {
+export class UnfinalizedBlocksService extends BaseUnfinalizedBlocksService<EthereumBlock> {
+  private supportsFinalization?: boolean;
+  private startupCheck = true;
+
   constructor(
     private readonly apiService: ApiService,
     nodeConfig: NodeConfig,
     storeCache: StoreCacheService,
   ) {
-    super(nodeConfig, storeCache);
+    super(new EthereumNodeConfig(nodeConfig), storeCache);
+  }
+
+  /**
+   * @param reindex - the function to reindex back before a fork
+   * @param supportsFinalization - If the chain supports the 'finalized' block tag this should be true.
+   * */
+  async init(
+    reindex: (targetHeight: number) => Promise<void>,
+    supportsFinalisation?: boolean,
+  ): Promise<number | undefined> {
+    this.supportsFinalization = supportsFinalisation;
+    return super.init(reindex);
+  }
+
+  // Detect a fork by walking back through unfinalized blocks
+  @profiler()
+  protected async hasForked(): Promise<Header | undefined> {
+    if (this.supportsFinalization) {
+      return super.hasForked();
+    }
+
+    // Startup check helps speed up finding a fork by checking the hash of the last unfinalized block
+    if (this.startupCheck) {
+      this.startupCheck = false;
+      const lastUnfinalized = last(this.unfinalizedBlocks);
+      if (lastUnfinalized) {
+        const checkUnfinalized = await this.getHeaderForHeight(
+          lastUnfinalized.blockHeight,
+        );
+
+        if (lastUnfinalized.blockHash !== checkUnfinalized.blockHash) {
+          return checkUnfinalized;
+        }
+      }
+    }
+
+    if (this.unfinalizedBlocks.length <= 2) {
+      return;
+    }
+
+    const i = this.unfinalizedBlocks.length - 1;
+    const current = this.unfinalizedBlocks[i];
+    const parent = this.unfinalizedBlocks[i - 1];
+
+    if (current.parentHash !== parent.blockHash) {
+      // We've found a fork now we need to find where the fork happened
+      logger.warn(
+        `Block fork detected at ${current.blockHeight}. Parent hash ${current.parentHash} doesn't match indexed parent ${parent.blockHash}.`,
+      );
+
+      return current;
+    }
+
+    return;
+  }
+
+  protected async getLastCorrectFinalizedBlock(
+    forkedHeader: Header,
+  ): Promise<number | undefined> {
+    if (this.supportsFinalization) {
+      return super.getLastCorrectFinalizedBlock(forkedHeader);
+    }
+
+    const bestVerifiableBlocks = this.unfinalizedBlocks.filter(
+      ({ blockHeight }) => blockHeight < forkedHeader.blockHeight,
+    );
+
+    let checkingHeader = forkedHeader;
+
+    // Work backwards through the blocks until we find a matching hash
+    for (const { blockHash, blockHeight } of bestVerifiableBlocks.reverse()) {
+      if (
+        blockHash === checkingHeader.blockHash ||
+        blockHash === checkingHeader.parentHash
+      ) {
+        return blockHeight;
+      }
+
+      // Get the new parent
+      checkingHeader = await this.getHeaderForHash(checkingHeader.parentHash);
+    }
+
+    try {
+      const poiHeader = await this.findFinalizedUsingPOI(checkingHeader);
+      logger.info(`Using POI to rewind to ${JSON.stringify(poiHeader)}`);
+      return poiHeader.blockHeight;
+    } catch (e) {
+      if (e.message === POI_NOT_ENABLED_ERROR_MESSAGE) {
+        console.log(
+          'HERE',
+          (this.nodeConfig as EthereumNodeConfig).blockForkReindex,
+          forkedHeader.blockHeight,
+        );
+        return Math.max(
+          0,
+          forkedHeader.blockHeight -
+            (this.nodeConfig as EthereumNodeConfig).blockForkReindex,
+        );
+      }
+      // TODO rewind back 1000+ blocks
+      logger.info('Failed to use POI to rewind block');
+      throw e;
+    }
   }
 
   @mainThreadOnly()

packages/node/src/yargs.ts

@@ -25,10 +25,17 @@ export const yargsOptions = yargsBuilder({
   runOptions: {
     'block-confirmations': {
       demandOption: false,
-      default: 20,
+      default: 200,
       describe:
-        'The number of blocks behind the head to be considered finalized, this has no effect with Ethereum',
+        'The number of blocks behind the head to be considered finalized for networks without deterministic finalisation such as Polygon POS',
       type: 'number',
     },
+    'block-fork-reindex': {
+      demandOption: false,
+      default: 1000,
+      type: 'number',
+      describe:
+        'The number of blocks to reindex if a fork happens before cached unfinalized blocks and POI is not enabled.',
+    },
   },
 });