fix(node-core): rewind a fork found while registering an unfinalized block

packages/node-core/CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Dictionary v1 queries kept every parsed query document in memory: each batch's query text is new, and both `graphql-tag`'s cache and Apollo Client's `QueryManager.transformCache` retained it, so a long sync's memory grew without bound. Queries are now posted as plain GraphQL over HTTP and nothing is parsed or cached (#3048)
 - Dictionary v1 queries threw a `TypeError` on a `null` body, `{data: null}` or `errors` that isn't an array, and their errors dropped the response body, so the `distinct` fallback missed non-standard error responses. The error now keeps the body and includes it in its message (#3051)
 - A timed-out dictionary v1 query kept running in the background with its connection and response body. It is now aborted a second after the query times out (#3051)
+- Rewinding a fork found while registering an unfinalized block failed the assertion `Expect best header and checking header to be at the same height` whenever a stored block was at or below the finalized height, so the node exited instead of rewinding. A fork found while registering now only signals a rewind: the stored blocks at or below the finalized height are checked against the finalized chain as for any other block, and the highest of them that matches, or the finalized block, is the rewind target. This also stops a stored fork block being chosen as the target when the backfill reports the fork (#3052)
 
 ### Changed
 - `DictionaryV1` sends queries with a protected `query()` method instead of an Apollo `client`, and `@apollo/client` is no longer a dependency (#3048)

packages/node-core/src/indexer/project.service.spec.ts

@@ -474,8 +474,8 @@ describe('BaseProjectService', () => {
       await setupProject(
         95,
         [
-          {blockHeight: 100, blockHash: 'a100', parentHash: 'a99', timestamp: new Date()},
           {blockHeight: 99, blockHash: 'a99', parentHash: 'a98', timestamp: new Date()},
+          {blockHeight: 100, blockHash: 'a100', parentHash: 'a99', timestamp: new Date()},
         ],
         90
       );

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

@@ -16,6 +16,8 @@ import {
  * If they are forked they will have an `f` at the end
  */
 let mockBestHeight = 150;
+// Headers off the canonical chain, by hash. Any other hash is canonical
+let forkHeaders: Record<string, Header> = {};
 
 const BlockchainService = {
   async getFinalizedHeader(): Promise<Header> {
@@ -27,6 +29,9 @@ const BlockchainService = {
     });
   },
   async getHeaderForHash(hash: string): Promise<Header> {
+    if (forkHeaders[hash]) {
+      return Promise.resolve(forkHeaders[hash]);
+    }
     const num = Number(hash.toString().replace('0xabc', '').replace('f', ''));
     return Promise.resolve({
       blockHeight: num,
@@ -85,6 +90,7 @@ describe('UnfinalizedBlocksService', () => {
 
   beforeEach(async () => {
     mockBestHeight = 150;
+    forkHeaders = {};
     unfinalizedBlocksService = new UnfinalizedBlocksService(
       {unfinalizedBlocks: true} as any,
       mockStoreCache(),
@@ -198,8 +204,23 @@ describe('UnfinalizedBlocksService', () => {
     await unfinalizedBlocksService.processUnfinalizedBlocks(mockBlock(115, '0xabc115', '0xabc114'));
     const res = await unfinalizedBlocksService.processUnfinalizedBlocks(mockBlock(116, '0xabc116', '0xabc115f'));
 
+    // The finalized block, as no stored block is at or below it
+    expect(res).toMatchObject({blockHash: '0xabcd', blockHeight: 110});
+  });
+
+  it('can handle a fork when latest unfinalized block has different parent and a block is finalized', async () => {
+    unfinalizedBlocksService.registerFinalizedBlock(mockBlock(110, '0xabcd').block.header);
+
+    await unfinalizedBlocksService.processUnfinalizedBlocks(mockBlock(111, '0xabc111f', '0xabc110f'));
+    // Forked block
+    await unfinalizedBlocksService.processUnfinalizedBlocks(mockBlock(112, '0xabc112', '0xabc111f'));
+
+    unfinalizedBlocksService.registerFinalizedBlock(mockBlock(111, '0xabc111f', '0xabc110f').block.header);
+
+    const res = await unfinalizedBlocksService.processUnfinalizedBlocks(mockBlock(113, '0xabc113f', '0xabc112f'));
+
     // Last valid block
-    expect(res).toMatchObject({blockHash: '0xabc110f', blockHeight: 110, parentHash: '0xabc109f'});
+    expect(res).toMatchObject({blockHash: '0xabc111f', blockHeight: 111});
   });
 
   it('can handle a fork when all unfinalized blocks are invalid', async () => {
@@ -496,6 +517,57 @@ describe('UnfinalizedBlocksService', () => {
       expect(result?.blockHeight).toBe(110);
     });
 
+    it('detects fork after backfill when new block has wrong parentHash and a block is finalized', async () => {
+      unfinalizedBlocksService.registerFinalizedBlock(mockBlock(110, '0xabcd').block.header);
+
+      await unfinalizedBlocksService.processUnfinalizedBlocks(mockBlock(111, '0xabc111f', '0xabc110f'));
+
+      unfinalizedBlocksService.registerFinalizedBlock(mockBlock(111, '0xabc111f', '0xabc110f').block.header);
+
+      // Register block 115 with wrong parentHash (should connect to 0xabc114f)
+      const result = await unfinalizedBlocksService.processUnfinalizedBlocks(mockBlock(115, '0xabc115', '0xwrong'));
+
+      expect(result).toMatchObject({blockHash: '0xabc111f', blockHeight: 111});
+    });
+
+    it('does not rewind to a stored fork block when the backfill finds the fork', async () => {
+      // 111x and 112x branch off the canonical 110
+      const fork111 = mockBlock(111, '0xabc111x', '0xabc110f');
+      const fork112 = mockBlock(112, '0xabc112x', '0xabc111x');
+      forkHeaders = {'0xabc111x': fork111.getHeader(), '0xabc112x': fork112.getHeader()};
+
+      unfinalizedBlocksService.registerFinalizedBlock(mockBlock(110, '0xabcd').block.header);
+
+      await unfinalizedBlocksService.processUnfinalizedBlocks(fork111);
+      await unfinalizedBlocksService.processUnfinalizedBlocks(fork112);
+
+      unfinalizedBlocksService.registerFinalizedBlock(mockBlock(111, '0xabc111f', '0xabc110f').block.header);
+
+      // Backfilling 113 finds that its parent is not the stored 112x
+      const result = await unfinalizedBlocksService.processUnfinalizedBlocks(mockBlock(115, '0xabc115f', '0xabc114f'));
+
+      expect(result).toMatchObject({blockHash: '0xabc110f', blockHeight: 110});
+    });
+
+    it('does not rewind to a stored fork block at the height the backfill reports', async () => {
+      // 111x and 112x branch off the canonical 110
+      const fork111 = mockBlock(111, '0xabc111x', '0xabc110f');
+      const fork112 = mockBlock(112, '0xabc112x', '0xabc111x');
+      forkHeaders = {'0xabc111x': fork111.getHeader(), '0xabc112x': fork112.getHeader()};
+
+      unfinalizedBlocksService.registerFinalizedBlock(mockBlock(110, '0xabcd').block.header);
+
+      await unfinalizedBlocksService.processUnfinalizedBlocks(fork111);
+      await unfinalizedBlocksService.processUnfinalizedBlocks(fork112);
+
+      unfinalizedBlocksService.registerFinalizedBlock(mockBlock(113, '0xabc113f', '0xabc112f').block.header);
+
+      // Backfilling 113 finds that its parent is not the stored 112x, which is below the finalized height
+      const result = await unfinalizedBlocksService.processUnfinalizedBlocks(mockBlock(115, '0xabc115f', '0xabc114f'));
+
+      expect(result).toMatchObject({blockHash: '0xabc110f', blockHeight: 110});
+    });
+
     it('throws error when block fetch fails during backfill', async () => {
       // Create a blockchain service that fails to fetch block 113
       const customBlockchainService = {

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

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: GPL-3.0
 
 import assert from 'assert';
+import util from 'node:util';
 import {Inject, Injectable} from '@nestjs/common';
 import {OnEvent} from '@nestjs/event-emitter';
 import {Transaction} from '@subql/x-sequelize';
@@ -15,7 +16,6 @@ import {mainThreadOnly} from '../utils';
 import {ProofOfIndex} from './entities';
 import {PoiBlock} from './poi';
 import {IStoreModelProvider} from './storeModelProvider';
-import util from 'node:util';
 
 const logger = getLogger('UnfinalizedBlocks');
 
@@ -104,23 +104,23 @@ export class UnfinalizedBlocksService<B = any> implements IUnfinalizedBlocksServ
 
   // If not for workers this could be private
   async processUnfinalizedBlockHeader(header?: Header): Promise<Header | undefined> {
-    let forkedHeader;
-    if (header) {
-      forkedHeader = await this.registerUnfinalizedBlock(header);
-    }
-    if (!forkedHeader) {
-      forkedHeader = await this.hasForked();
-    }
+    // A fork found while registering only signals a rewind. It can be reported from above the
+    // finalized height or from a stored fork block, so hasForked decides where to rewind to
+    const registrationFork = header ? await this.registerUnfinalizedBlock(header) : undefined;
 
-    if (!forkedHeader) {
-      // Remove blocks that are now confirmed finalized
-      await this.deleteFinalizedBlock();
-    } else {
+    const forkedHeader = await this.hasForked();
+    if (forkedHeader) {
       // Get the last unfinalized block that is now finalized
       return this.getLastCorrectFinalizedBlock(forkedHeader);
     }
 
-    return;
+    if (registrationFork) {
+      // The fork is above the finalized height, and hasForked has verified the highest stored block at or below it
+      return this.getClosestRecord(this.finalizedBlockNumber) ?? this.finalizedHeader;
+    }
+
+    // Remove blocks that are now confirmed finalized
+    await this.deleteFinalizedBlock();
   }
 
   async processUnfinalizedBlocks(block: IBlock<B>): Promise<Header | undefined> {