New fork detection for probabilistic finalization chains

packages/node/CHANGELOG.md

@@ -5,6 +5,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
+### Added
+- Different method for detecting block forks for chains with probabalistic finalization (#217)
 
 ## [3.3.6] - 2023-11-23
 ### Fixed

packages/node/package.json

@@ -27,7 +27,7 @@
     "@nestjs/schedule": "^3.0.1",
     "@subql/common": "^3.3.0",
     "@subql/common-ethereum": "workspace:*",
-    "@subql/node-core": "^6.4.2",
+    "@subql/node-core": "^7.0.0",
     "@subql/testing": "^2.0.2",
     "@subql/types-ethereum": "workspace:*",
     "cacheable-lookup": "6",

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';
@@ -99,7 +98,11 @@ export class EthereumApi implements ApiWrapper {
   private name: string;
 
   // Ethereum POS
-  private supportsFinalization = true;
+  private _supportsFinalization = true;
+
+  get supportsFinalization(): boolean {
+    return this._supportsFinalization;
+  }
 
   /**
    * @param {string} endpoint - The endpoint of the RPC provider
@@ -164,7 +167,7 @@ export class EthereumApi implements ApiWrapper {
         ]);
 
       this.genesisBlock = genesisBlock;
-      this.supportsFinalization = supportsFinalization && supportsSafe;
+      this._supportsFinalization = supportsFinalization && supportsSafe;
       this.chainId = network.chainId;
       this.name = network.name;
     } catch (e) {

packages/node/src/ethereum/ethers/celo/celo-json-rpc-batch-provider.spec.ts

@@ -30,6 +30,6 @@ describe('CeloJsonRpcProvider', () => {
     const block = formatBlock(
       await provider.send('eth_getBlockByNumber', ['latest', true]),
     );
-    expect(BigNumber.from(block.gasLimit)).toEqual(BigNumber.from(0x01e84800));
+    expect(BigNumber.from(block.gasLimit).gte(constants.Zero)).toBeTruthy();
   });
 });

packages/node/src/ethereum/ethers/celo/celo-json-rpc-provider.spec.ts

@@ -28,6 +28,6 @@ describe('CeloJsonRpcProvider', () => {
     const block = formatBlock(
       await provider.send('eth_getBlockByNumber', ['latest', true]),
     );
-    expect(BigNumber.from(block.gasLimit)).toEqual(BigNumber.from(0x01e84800));
+    expect(BigNumber.from(block.gasLimit).gte(constants.Zero)).toBeTruthy();
   });
 });

packages/node/src/ethereum/ethers/celo/celo-ws-provider.spec.ts

@@ -28,6 +28,6 @@ describe('CeloJsonRpcProvider', () => {
     const block = formatBlock(
       await provider.send('eth_getBlockByNumber', ['latest', true]),
     );
-    expect(BigNumber.from(block.gasLimit)).toEqual(BigNumber.from(0x01e84800));
+    expect(BigNumber.from(block.gasLimit).gte(constants.Zero)).toBeTruthy();
   });
 });

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/project.service.ts

@@ -29,7 +29,8 @@ const { version: packageVersion } = require('../../package.json');
 @Injectable()
 export class ProjectService extends BaseProjectService<
   EthereumApiService,
-  EthereumProjectDs
+  EthereumProjectDs,
+  UnfinalizedBlocksService
 > {
   protected packageVersion = packageVersion;
 
@@ -75,4 +76,11 @@ export class ProjectService extends BaseProjectService<
     // TODO update this when implementing skipBlock feature for Eth
     this.apiService.updateBlockFetching();
   }
+
+  protected async initUnfinalized(): Promise<number | undefined> {
+    return this.unfinalizedBlockService.init(
+      this.reindex.bind(this),
+      this.apiService.api.supportsFinalization,
+    );
+  }
 }

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

@@ -0,0 +1,214 @@
+// Copyright 2020-2023 SubQuery Pte Ltd authors & contributors
+// SPDX-License-Identifier: GPL-3.0
+
+import { hexZeroPad } from '@ethersproject/bytes';
+import {
+  ApiService,
+  CacheMetadataModel,
+  Header,
+  NodeConfig,
+  PoiBlock,
+  StoreCacheService,
+  METADATA_UNFINALIZED_BLOCKS_KEY,
+  METADATA_LAST_FINALIZED_PROCESSED_KEY,
+} from '@subql/node-core';
+import { EthereumNodeConfig } from '../configure/NodeConfig';
+import { UnfinalizedBlocksService } from './unfinalizedBlocks.service';
+
+// Adds 0 padding so we can convert to POI block
+const hexify = (input: string) => hexZeroPad(input, 4);
+
+const makeHeader = (height: number, finalized?: boolean): Header => ({
+  blockHeight: height,
+  blockHash: hexify(`0xABC${height}${finalized ? 'f' : ''}`),
+  parentHash: hexify(`0xABC${height - 1}${finalized ? 'f' : ''}`),
+});
+
+const getMockApi = (): ApiService => {
+  return {
+    api: {
+      getBlockByHeightOrHash: (hash: string | number) => {
+        const num =
+          typeof hash === 'number'
+            ? hash
+            : Number(
+                hash
+                  .toString()
+                  .replace('0x', '')
+                  .replace('ABC', '')
+                  .replace('f', ''),
+              );
+        return Promise.resolve({
+          number: num,
+          hash: typeof hash === 'number' ? hexify(`0xABC${hash}f`) : hash,
+          parentHash: hexify(`0xABC${num - 1}f`),
+        });
+      },
+      getFinalizedBlock: jest.fn(() => ({
+        number: 110,
+        hash: '0xABC110f',
+        parentHash: '0xABC109f',
+      })),
+    },
+  } as any;
+};
+
+function getMockMetadata(): any {
+  const data: Record<string, any> = {};
+  return {
+    upsert: ({ key, value }: any) => (data[key] = value),
+    findOne: ({ where: { key } }: any) => ({ value: data[key] }),
+    findByPk: (key: string) => data[key],
+    find: (key: string) => data[key],
+  } as any;
+}
+
+function mockStoreCache(): StoreCacheService {
+  return {
+    metadata: new CacheMetadataModel(getMockMetadata()),
+    poi: {
+      getPoiBlocksBefore: jest.fn(() => [
+        PoiBlock.create(99, hexify('0xABC99f'), new Uint8Array(), ''),
+      ]),
+    },
+  } as any as StoreCacheService;
+}
+
+describe('UnfinalizedBlockService', () => {
+  let unfinalizedBlocks: UnfinalizedBlocksService;
+  let storeCache: StoreCacheService;
+
+  beforeEach(() => {
+    storeCache = mockStoreCache();
+
+    unfinalizedBlocks = new UnfinalizedBlocksService(
+      getMockApi(),
+      new NodeConfig({
+        unfinalizedBlocks: true,
+        blockForkReindex: 1000,
+      } as any) as EthereumNodeConfig,
+      storeCache,
+    );
+  });
+
+  it('handles a block fork', async () => {
+    await unfinalizedBlocks.init(jest.fn());
+
+    (unfinalizedBlocks as any)._unfinalizedBlocks = [
+      makeHeader(100),
+      makeHeader(101),
+      makeHeader(102),
+      makeHeader(103, true), // Where the fork started
+      makeHeader(104),
+      makeHeader(105),
+      makeHeader(106),
+      makeHeader(107),
+      makeHeader(108),
+      makeHeader(109),
+      makeHeader(110),
+    ];
+
+    const rewind = await unfinalizedBlocks.processUnfinalizedBlockHeader(
+      makeHeader(111, true),
+    );
+
+    expect(rewind).toEqual(103);
+  });
+
+  it('uses POI blocks if there are not enough cached unfinalized blocks', async () => {
+    await unfinalizedBlocks.init(jest.fn());
+
+    (unfinalizedBlocks as any)._unfinalizedBlocks = [
+      makeHeader(100),
+      makeHeader(101),
+      makeHeader(102),
+      makeHeader(103),
+      makeHeader(104),
+      makeHeader(105),
+      makeHeader(106),
+      makeHeader(107),
+      makeHeader(108),
+      makeHeader(109),
+      makeHeader(110),
+    ];
+
+    const spy = jest.spyOn(storeCache.poi as any, 'getPoiBlocksBefore');
+
+    const rewind = await unfinalizedBlocks.processUnfinalizedBlockHeader(
+      makeHeader(111, true),
+    );
+
+    expect(rewind).toEqual(99);
+    expect(spy).toHaveBeenCalled();
+  });
+
+  // The finalized block is after the cached unfinalized blocks, they should be rechecked
+  it('startup, correctly checks for forks after cached unfinalized blocks', async () => {
+    storeCache.metadata.set(
+      METADATA_UNFINALIZED_BLOCKS_KEY,
+      JSON.stringify(<Header[]>[
+        makeHeader(99, true),
+        makeHeader(100),
+        makeHeader(101),
+      ]),
+    );
+
+    storeCache.metadata.set(METADATA_LAST_FINALIZED_PROCESSED_KEY, 99);
+
+    const rewind = jest.fn();
+
+    await unfinalizedBlocks.init(rewind);
+
+    // It should fall back to poi in this case
+    expect(rewind).toHaveBeenCalledWith(99);
+  });
+
+  it('startup, correctly checks for forks within cached unfinalized blocks', async () => {
+    storeCache.metadata.set(
+      METADATA_UNFINALIZED_BLOCKS_KEY,
+      JSON.stringify(<Header[]>[
+        makeHeader(110),
+        makeHeader(111),
+        makeHeader(112),
+      ]),
+    );
+
+    storeCache.metadata.set(METADATA_LAST_FINALIZED_PROCESSED_KEY, 109);
+
+    const rewind = jest.fn();
+
+    await unfinalizedBlocks.init(rewind);
+
+    // It should fall back to poi in this case
+    expect(rewind).toHaveBeenCalledWith(99);
+  });
+
+  it('doesnt throw if there are no unfinalized blocks on startup', async () => {
+    storeCache.metadata.set(METADATA_LAST_FINALIZED_PROCESSED_KEY, 109);
+
+    await expect(unfinalizedBlocks.init(jest.fn())).resolves.not.toThrow();
+  });
+
+  it('rewinds using blockForkReindex value if poi is not enabled', async () => {
+    // Do this to "disable" poi
+    (storeCache as any).poi = null;
+
+    storeCache.metadata.set(
+      METADATA_UNFINALIZED_BLOCKS_KEY,
+      JSON.stringify(<Header[]>[
+        makeHeader(110),
+        makeHeader(111),
+        makeHeader(112),
+      ]),
+    );
+
+    storeCache.metadata.set(METADATA_LAST_FINALIZED_PROCESSED_KEY, 109);
+
+    const rewind = jest.fn();
+
+    await unfinalizedBlocks.init(rewind);
+
+    // It should fall back to poi in this case
+    expect(rewind).toHaveBeenCalledWith(0);
+  });
+});