Handle Solana skipped slots in block fetching and unfinalized block t…

packages/node/package.json

@@ -25,6 +25,7 @@
     "@nestjs/event-emitter": "^3.0.0",
     "@nestjs/platform-express": "^11.0.8",
     "@nestjs/schedule": "^5.0.1",
+    "@solana/errors": "^3.0.2",
     "@solana/kit": "^3.0.2",
     "@subql/common-solana": "workspace:*",
     "@subql/node-core": "^18.3.3",

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

@@ -29,41 +29,69 @@ const headerFromHeight = (
   height: number,
   finalized = false,
   parentFinalized = false,
+  parentHeight = height - 1,
 ): Header => ({
   blockHeight: height,
   blockHash: `0x${height}${finalized ? 'f' : ''}`,
-  parentHash: `0x${height - 1}${parentFinalized ? 'f' : ''}`,
+  parentHash: `0x${parentHeight}${parentFinalized ? 'f' : ''}`,
   timestamp: new Date('2025-08-27T23:07:53.486Z'),
 });
 
+const getPreviousProducedSlot = (
+  height: number,
+  skippedSlots: Set<number>,
+): number => {
+  let parentHeight = height - 1;
+  while (skippedSlots.has(parentHeight)) {
+    parentHeight--;
+  }
+  return parentHeight;
+};
+
 const getMockBlockchainService = (
   finalizedHeight = 100,
   skippedSlots: number[] = [],
 ): IBlockchainService & {
   setFinalizedHeight: (newHeight: number) => void;
+  setForkedParent: (height: number, parentHash: string) => void;
   setSkippedSlots: (slots: number[]) => void;
 } => {
   let _finalizedHeight = finalizedHeight;
   let _skippedSlots = new Set(skippedSlots);
+  const _forkedParents = new Map<number, string>();
 
   return {
     getFinalizedHeader: () =>
-      Promise.resolve(headerFromHeight(_finalizedHeight, true)),
+      Promise.resolve(
+        headerFromHeight(
+          _finalizedHeight,
+          true,
+          true,
+          getPreviousProducedSlot(_finalizedHeight, _skippedSlots),
+        ),
+      ),
     getHeaderForHeight: (height: number) => {
       // Same behaviour as in SolanaApi
       if (_skippedSlots.has(height)) {
         // No block for that slot
         throw new BlockUnavailableError();
       }
-      return Promise.resolve(
-        headerFromHeight(
-          height,
-          height <= _finalizedHeight,
-          height < _finalizedHeight,
-        ),
+      const parentHeight = getPreviousProducedSlot(height, _skippedSlots);
+      const header = headerFromHeight(
+        height,
+        height <= _finalizedHeight,
+        parentHeight < _finalizedHeight,
+        parentHeight,
       );
+      const forkedParent = _forkedParents.get(height);
+      if (forkedParent) {
+        header.parentHash = forkedParent;
+      }
+      return Promise.resolve(header);
     },
     setFinalizedHeight: (newHeight: number) => (_finalizedHeight = newHeight),
+    setForkedParent: (height: number, parentHash: string) =>
+      _forkedParents.set(height, parentHash),
     setSkippedSlots: (slots: number[]) => (_skippedSlots = new Set(slots)),
   } as any;
 };
@@ -105,7 +133,7 @@ describe('Unfinalized blocks', () => {
 
   it('handles block forks when there are missed slots', async () => {
     const store = getMockStoreModelProvider();
-    const blockchain = getMockBlockchainService(100);
+    const blockchain = getMockBlockchainService(100, [103]);
     const unfinalizedBlocks = new UnfinalizedBlocksService(
       new NodeConfig({} as any),
       store,
@@ -119,14 +147,23 @@ describe('Unfinalized blocks', () => {
     const forkHeight = 108;
     while (height <= 110) {
       if (height === forkHeight) {
-        blockchain.setSkippedSlots([103]);
         blockchain.setFinalizedHeight(forkHeight);
         unfinalizedBlocks.registerFinalizedBlock(
-          headerFromHeight(forkHeight, true, true),
+          await blockchain.getHeaderForHeight(forkHeight),
         );
       }
+      let header: Header;
+      try {
+        header = await blockchain.getHeaderForHeight(height);
+      } catch (e) {
+        if (e instanceof BlockUnavailableError) {
+          height++;
+          continue;
+        }
+        throw e;
+      }
       const rewindTo = await unfinalizedBlocks.processUnfinalizedBlockHeader(
-        await blockchain.getHeaderForHeight(height),
+        header,
       );
       if (rewindTo) {
         reindex(rewindTo);
@@ -137,4 +174,74 @@ describe('Unfinalized blocks', () => {
 
     expect(reindex).toHaveBeenCalledWith(headerFromHeight(100, true, true));
   });
+
+  it('rebuilds the parent chain across actual and skipped slots', async () => {
+    const store = getMockStoreModelProvider();
+    const blockchain = getMockBlockchainService(100, [103]);
+    const unfinalizedBlocks = new UnfinalizedBlocksService(
+      new NodeConfig({} as any),
+      store,
+      blockchain,
+    );
+
+    await unfinalizedBlocks.init(jest.fn());
+
+    await unfinalizedBlocks.processUnfinalizedBlockHeader(
+      await blockchain.getHeaderForHeight(101),
+    );
+    await unfinalizedBlocks.processUnfinalizedBlockHeader(
+      await blockchain.getHeaderForHeight(105),
+    );
+
+    expect((unfinalizedBlocks as any).unfinalizedBlocks).toMatchObject([
+      await blockchain.getHeaderForHeight(101),
+      await blockchain.getHeaderForHeight(102),
+      await blockchain.getHeaderForHeight(104),
+      await blockchain.getHeaderForHeight(105),
+    ]);
+  });
+
+  it('rolls back when a backfilled slot has the wrong parent hash', async () => {
+    const store = getMockStoreModelProvider();
+    const blockchain = getMockBlockchainService(100);
+    const unfinalizedBlocks = new UnfinalizedBlocksService(
+      new NodeConfig({} as any),
+      store,
+      blockchain,
+    );
+
+    await unfinalizedBlocks.init(jest.fn());
+    await unfinalizedBlocks.processUnfinalizedBlockHeader(
+      await blockchain.getHeaderForHeight(101),
+    );
+
+    blockchain.setForkedParent(102, '0xfork');
+    const rewindTo = await unfinalizedBlocks.processUnfinalizedBlockHeader(
+      await blockchain.getHeaderForHeight(105),
+    );
+
+    expect(rewindTo).toMatchObject(headerFromHeight(100, true, true));
+  });
+
+  it('rolls back when a new block does not connect after skipped slots', async () => {
+    const store = getMockStoreModelProvider();
+    const blockchain = getMockBlockchainService(100, [103]);
+    const unfinalizedBlocks = new UnfinalizedBlocksService(
+      new NodeConfig({} as any),
+      store,
+      blockchain,
+    );
+
+    await unfinalizedBlocks.init(jest.fn());
+    await unfinalizedBlocks.processUnfinalizedBlockHeader(
+      await blockchain.getHeaderForHeight(101),
+    );
+
+    blockchain.setForkedParent(105, '0xfork');
+    const rewindTo = await unfinalizedBlocks.processUnfinalizedBlockHeader(
+      await blockchain.getHeaderForHeight(105),
+    );
+
+    expect(rewindTo).toMatchObject(headerFromHeight(100, true, true));
+  });
 });

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

@@ -121,11 +121,13 @@ export class UnfinalizedBlocksService<B = any>
   async processUnfinalizedBlockHeader(
     header?: Header,
   ): Promise<Header | undefined> {
+    let forkedHeader: Header | undefined;
+
     if (header) {
-      await this.registerUnfinalizedBlock(header);
+      forkedHeader = await this.registerUnfinalizedBlock(header);
     }
 
-    const forkedHeader = await this.hasForked();
+    forkedHeader ??= await this.hasForked();
 
     if (!forkedHeader) {
       // Remove blocks that are now confirmed finalized
@@ -156,23 +158,84 @@ export class UnfinalizedBlocksService<B = any>
     this._finalizedHeader = header;
   }
 
-  private async registerUnfinalizedBlock(header: Header): Promise<void> {
+  private async registerUnfinalizedBlock(
+    header: Header,
+  ): Promise<Header | undefined> {
     if (header.blockHeight <= this.finalizedBlockNumber) return;
 
-    // Ensure order
-    const lastUnfinalizedHeight = last(this.unfinalizedBlocks)?.blockHeight;
+    const lastUnfinalized = last(this.unfinalizedBlocks);
+    const lastUnfinalizedHeight = lastUnfinalized?.blockHeight;
     if (
       lastUnfinalizedHeight !== undefined &&
-      lastUnfinalizedHeight + 1 !== header.blockHeight
+      lastUnfinalizedHeight >= header.blockHeight
     ) {
       exitWithError(
         `Unfinalized block is not sequential, lastUnfinalizedBlock='${lastUnfinalizedHeight}', newUnfinalizedBlock='${header.blockHeight}'`,
         logger,
       );
     }
 
+    if (!lastUnfinalized) {
+      this.unfinalizedBlocks.push(header);
+      await this.saveUnfinalizedBlocks(this.unfinalizedBlocks);
+      return;
+    }
+
+    const lastProducedHeight = lastUnfinalized.blockHeight;
+    if (lastProducedHeight + 1 !== header.blockHeight) {
+      const forkedHeader = await this.backfillSkippedSlots(
+        lastProducedHeight + 1,
+        header.blockHeight - 1,
+      );
+
+      if (forkedHeader) {
+        return forkedHeader;
+      }
+    }
+
+    const latestUnfinalized = last(this.unfinalizedBlocks);
+    if (
+      latestUnfinalized &&
+      header.parentHash !== latestUnfinalized.blockHash
+    ) {
+      logger.warn(
+        `Block fork found, enqueued un-finalized block at ${header.blockHeight} with parent hash ${header.parentHash}, expected parent hash is ${latestUnfinalized.blockHash}.`,
+      );
+      return header;
+    }
+
     this.unfinalizedBlocks.push(header);
     await this.saveUnfinalizedBlocks(this.unfinalizedBlocks);
+    return;
+  }
+
+  private async backfillSkippedSlots(
+    startHeight: number,
+    endHeight: number,
+  ): Promise<Header | undefined> {
+    for (let height = startHeight; height <= endHeight; height++) {
+      let header: Header;
+      try {
+        header = await this.blockchainService.getHeaderForHeight(height);
+      } catch (e) {
+        if (e instanceof BlockUnavailableError) {
+          continue;
+        }
+        throw e;
+      }
+
+      const previousHeader = last(this.unfinalizedBlocks);
+      if (previousHeader && header.parentHash !== previousHeader.blockHash) {
+        logger.warn(
+          `Block fork found while rebuilding un-finalized chain at ${header.blockHeight} with parent hash ${header.parentHash}, expected parent hash is ${previousHeader.blockHash}.`,
+        );
+        return header;
+      }
+
+      this.unfinalizedBlocks.push(header);
+    }
+
+    return;
   }
 
   private async deleteFinalizedBlock(): Promise<void> {

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

@@ -21,6 +21,10 @@ import { BlockContent, getBlockSize } from '../types';
 
 export type FetchBlockResponse = Header;
 
+// The base class unwraps IBlock to BlockContent before calling toBlockResponse, losing the slot
+// that was used to fetch it. Track it here so the header reflects the requested slot, not parentSlot + 1.
+const blockSlots = new WeakMap<BlockContent, number>();
+
 export type WorkerStatusResponse = {
   threadId: number;
   isIndexing: boolean;
@@ -57,11 +61,12 @@ export class WorkerService extends BaseWorkerService<
     extra: {},
   ): Promise<IBlock<BlockContent>> {
     const [block] = await this.apiService.fetchBlocks([heights]);
+    blockSlots.set(block.block, heights);
     return block;
   }
 
   protected toBlockResponse(block: BlockContent): Header {
-    return solanaBlockToHeader(block);
+    return solanaBlockToHeader(block, blockSlots.get(block));
   }
 
   protected async processFetchedBlock(

packages/node/src/solana/api.solana.spec.ts

@@ -21,9 +21,7 @@ import {
 
 // Add api key to work
 const HTTP_ENDPOINT =
-  process.env.HTTP_ENDPOINT ??
-  'https://api.mainnet-beta.solana.com' ??
-  'https://solana.api.onfinality.io/public';
+  process.env.HTTP_ENDPOINT ?? 'https://api.mainnet-beta.solana.com';
 
 const IDL_swap: IdlV01 = require('../../test/swapFpHZwjELNnjvThjajtiVmkz3yPQEHjLtka2fwHW.idl.json');