Implement ethereum specific fork detection and POI fallback

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

@@ -10,8 +10,15 @@ import {
   mainThreadOnly,
   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 {
@@ -23,6 +30,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,
@@ -31,6 +40,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;
+  }
+
   @mainThreadOnly()
   protected blockToHeader(block: BlockWrapper): Header {
     return blockToHeader(block.block);