Implement ethereum specific fork detection and POI fallback

packages/node/package.json

@@ -26,7 +26,7 @@
     "@nestjs/schedule": "^3.0.1",
     "@subql/common": "2.6.1-4",
     "@subql/common-ethereum": "workspace:*",
-    "@subql/node-core": "4.2.4-9",
+    "@subql/node-core": "4.2.4-10",
     "@subql/testing": "^2.0.0",
     "@subql/types": "^2.1.4",
     "@subql/types-ethereum": "workspace:*",

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

@@ -0,0 +1,134 @@
+// 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,
+} from '@subql/node-core';
+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(() => ({})),
+    },
+  } 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(),
+      { unfinalizedBlocks: true } as NodeConfig,
+      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();
+  });
+});

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

@@ -9,8 +9,15 @@ import {
   Header,
   NodeConfig,
   StoreCacheService,
+  getLogger,
+  ProofOfIndex,
+  PoiBlock,
+  profiler,
 } from '@subql/node-core';
 import { BlockWrapper, EthereumBlock } from '@subql/types-ethereum';
+import { isEqual } from 'lodash';
+
+const logger = getLogger('UnfinalizedBlocksService');
 
 export function blockToHeader(block: EthereumBlock | Block): Header {
   return {
@@ -22,6 +29,8 @@ export function blockToHeader(block: EthereumBlock | Block): Header {
 
 @Injectable()
 export class UnfinalizedBlocksService extends BaseUnfinalizedBlocksService<BlockWrapper> {
+  private supportsFinalization?: boolean;
+
   constructor(
     private readonly apiService: ApiService,
     nodeConfig: NodeConfig,
@@ -30,6 +39,101 @@ export class UnfinalizedBlocksService extends BaseUnfinalizedBlocksService<Block
     super(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();
+    }
+
+    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}.`,
+      );
+
+      let parentIndex = i - 1;
+      let indexedParent = parent;
+      let chainParent = await this.getHeaderForHash(current.parentHash);
+      while (chainParent.blockHash !== indexedParent.blockHash) {
+        parentIndex--;
+        // We've exhausted cached unfinalized blocks, we can check POI now for forks.
+        if (parentIndex < 0) {
+          const poiModel = this.storeCache.poi;
+          if (!poiModel) {
+            // TODO update message to explain how to recover from this.
+            throw new Error(
+              'Ran out of cached unfinalized blocks. Unable to find if a fork was indexed.',
+            );
+          }
+
+          logger.warn('Using POI to find older block fork');
+
+          const indexedBlocks: ProofOfIndex[] =
+            await poiModel.getPoiBlocksBefore(chainParent.blockHeight);
+
+          // Work backwards to find a block on chain that matches POI
+          for (const indexedBlock of indexedBlocks) {
+            const chainHeader = await this.getHeaderForHeight(indexedBlock.id);
+
+            // Need to convert to PoiBlock to encode block hash to Uint8Array properly
+            const testPoiBlock = PoiBlock.create(
+              chainHeader.blockHeight,
+              chainHeader.blockHash,
+              new Uint8Array(),
+              indexedBlock.projectId,
+            );
+
+            // Need isEqual because of Uint8Array type
+            if (
+              isEqual(testPoiBlock.chainBlockHash, indexedBlock.chainBlockHash)
+            ) {
+              return chainHeader;
+            }
+          }
+        }
+        indexedParent = this.unfinalizedBlocks[parentIndex];
+        chainParent = await this.getHeaderForHash(chainParent.parentHash);
+      }
+
+      return chainParent;
+    }
+
+    return;
+  }
+
+  // eslint-disable-next-line @typescript-eslint/require-await
+  protected async getLastCorrectFinalizedBlock(
+    forkedHeader: Header,
+  ): Promise<number | undefined> {
+    if (this.supportsFinalization) {
+      return super.getLastCorrectFinalizedBlock(forkedHeader);
+    }
+
+    // TODO update lastChecked block height to clean up unfinalized blocks
+    return forkedHeader.blockHeight;
+  }
+
   protected blockToHeader(block: BlockWrapper): Header {
     return blockToHeader(block.block);
   }

yarn.lock

@@ -3071,9 +3071,9 @@ __metadata:
   languageName: node
   linkType: hard
 
-"@subql/node-core@npm:4.2.4-9":
-  version: 4.2.4-9
-  resolution: "@subql/node-core@npm:4.2.4-9"
+"@subql/node-core@npm:4.2.4-10":
+  version: 4.2.4-10
+  resolution: "@subql/node-core@npm:4.2.4-10"
   dependencies:
     "@apollo/client": ^3.7.16
     "@nestjs/common": ^9.4.0
@@ -3098,7 +3098,7 @@ __metadata:
     tar: ^6.1.11
     vm2: ^3.9.19
     yargs: ^16.2.0
-  checksum: 5b48f3913c72dd6804ccb2d3bfc6af46ce30e49484cec5503e0e90485e97e59ae074c03ef28b7d6f4e4efeae22517662464f2e359f5623824e73beb1e5f92e40
+  checksum: 75407ef3007c48ea73db6b9cf0dc6379fe1052e4442586c3eaa5ba6caed3383d136ef8e7b3dfbf44688b2020c193b1d7dfff62cad6d4e8ea46d4946deb03a7e0
   languageName: node
   linkType: hard
 
@@ -3116,7 +3116,7 @@ __metadata:
     "@nestjs/testing": ^9.4.0
     "@subql/common": 2.6.1-4
     "@subql/common-ethereum": "workspace:*"
-    "@subql/node-core": 4.2.4-9
+    "@subql/node-core": 4.2.4-10
     "@subql/testing": ^2.0.0
     "@subql/types": ^2.1.4
     "@subql/types-ethereum": "workspace:*"