diff --git a/packages/node/package.json b/packages/node/package.json index 388e1eba7..75d5ef46f 100644 --- a/packages/node/package.json +++ b/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", diff --git a/packages/node/src/indexer/unfinalizedBlocks.service.spec.ts b/packages/node/src/indexer/unfinalizedBlocks.service.spec.ts index ec29d6af3..17617fc8f 100644 --- a/packages/node/src/indexer/unfinalizedBlocks.service.spec.ts +++ b/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 => { + 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(); 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)); + }); }); diff --git a/packages/node/src/indexer/unfinalizedBlocks.service.ts b/packages/node/src/indexer/unfinalizedBlocks.service.ts index ae01ccaa3..685beae6f 100644 --- a/packages/node/src/indexer/unfinalizedBlocks.service.ts +++ b/packages/node/src/indexer/unfinalizedBlocks.service.ts @@ -121,11 +121,13 @@ export class UnfinalizedBlocksService async processUnfinalizedBlockHeader( header?: Header, ): Promise
{ + 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,14 +158,16 @@ export class UnfinalizedBlocksService this._finalizedHeader = header; } - private async registerUnfinalizedBlock(header: Header): Promise { + private async registerUnfinalizedBlock( + header: Header, + ): Promise
{ 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}'`, @@ -171,8 +175,67 @@ export class UnfinalizedBlocksService ); } + 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
{ + 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 { diff --git a/packages/node/src/indexer/worker/worker.service.ts b/packages/node/src/indexer/worker/worker.service.ts index b39347ad0..5084ce961 100644 --- a/packages/node/src/indexer/worker/worker.service.ts +++ b/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(); + export type WorkerStatusResponse = { threadId: number; isIndexing: boolean; @@ -57,11 +61,12 @@ export class WorkerService extends BaseWorkerService< extra: {}, ): Promise> { 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( diff --git a/packages/node/src/solana/api.solana.spec.ts b/packages/node/src/solana/api.solana.spec.ts index 379166ccb..2f01590f3 100644 --- a/packages/node/src/solana/api.solana.spec.ts +++ b/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'); diff --git a/packages/node/src/solana/api.solana.ts b/packages/node/src/solana/api.solana.ts index bf3e8858f..0ce2c3d7b 100644 --- a/packages/node/src/solana/api.solana.ts +++ b/packages/node/src/solana/api.solana.ts @@ -3,6 +3,10 @@ import { EventEmitter2 } from '@nestjs/event-emitter'; import { assertIsAddress } from '@solana/addresses'; +import { + isSolanaError, + SOLANA_ERROR__JSON_RPC__SERVER_ERROR_SLOT_SKIPPED, +} from '@solana/errors'; import { createSolanaRpc, Rpc } from '@solana/rpc'; import { SolanaRpcApi } from '@solana/rpc-api'; import { @@ -30,6 +34,13 @@ export type SolanaSafeApi = undefined; const REQUEST_TIMEOUT = 30_000; +// Solana doesn't produce a block for every slot. For a skipped slot the RPC throws rather than returning null. +// Other "not available" RPC errors (e.g. block not yet rooted on this node, or missing from long-term storage) +// don't reliably mean the slot was skipped, so they're intentionally not treated as a skip here. +function isSkippedSlotError(e: unknown): boolean { + return isSolanaError(e, SOLANA_ERROR__JSON_RPC__SERVER_ERROR_SLOT_SKIPPED); +} + export class SolanaApi { #client: Rpc; @@ -154,21 +165,29 @@ export class SolanaApi { async getHeaderByHeight(height: number | bigint): Promise
{ const slot = typeof height === 'number' ? BigInt(height) : height; - const block = await this.#client - .getBlock(slot, { - maxSupportedTransactionVersion: 0, - transactionDetails: 'none', - rewards: false, - commitment: 'confirmed', // This is used by unfinalized blocks - }) - .send({ abortSignal: AbortSignal.timeout(this.#requestTimeout) }); + let block; + try { + block = await this.#client + .getBlock(slot, { + maxSupportedTransactionVersion: 0, + transactionDetails: 'none', + rewards: false, + commitment: 'confirmed', // This is used by unfinalized blocks + }) + .send({ abortSignal: AbortSignal.timeout(this.#requestTimeout) }); + } catch (e) { + if (isSkippedSlotError(e)) { + throw new BlockUnavailableError(); + } + throw e; + } if (!block) { // No block for that slot throw new BlockUnavailableError(); } - return solanaBlockToHeader(block); + return solanaBlockToHeader(block, Number(slot)); } async fetchBlock(blockNumber: number): Promise> { @@ -187,8 +206,14 @@ export class SolanaApi { } this.eventEmitter.emit('fetchBlock'); - return formatBlockUtil(transformBlock(rawBlock, this.decoder)); + return formatBlockUtil( + transformBlock(rawBlock, this.decoder), + blockNumber, + ); } catch (e: any) { + if (isSkippedSlotError(e)) { + throw new BlockUnavailableError(); + } console.log('Failed to fetch block', blockNumber, e.message); throw this.handleError(e); } diff --git a/packages/node/src/solana/block.solana.ts b/packages/node/src/solana/block.solana.ts index bcf63b290..8c5d7273e 100644 --- a/packages/node/src/solana/block.solana.ts +++ b/packages/node/src/solana/block.solana.ts @@ -265,16 +265,22 @@ export function transformBlock( export function formatBlockUtil( block: B, + slot?: number, ): IBlock { return { block, - getHeader: () => solanaBlockToHeader(block), + getHeader: () => solanaBlockToHeader(block, slot), }; } -export function solanaBlockToHeader(block: BaseSolanaBlock): Header { +export function solanaBlockToHeader( + block: BaseSolanaBlock, + slot?: number, +): Header { return { - blockHeight: Number(block.parentSlot) + 1, // The blocks don't include the slot because they assume you know that when making the request + // The response doesn't include its own slot, so callers that know the slot they requested + // should pass it in. Falling back to parentSlot + 1 is wrong when the previous slot was skipped. + blockHeight: slot ?? Number(block.parentSlot) + 1, blockHash: block.blockhash, parentHash: block.previousBlockhash, timestamp: new Date(Number(block.blockTime) * 1000), diff --git a/packages/node/src/solana/decoder.spec.ts b/packages/node/src/solana/decoder.spec.ts index e79d0a819..f897adf9b 100644 --- a/packages/node/src/solana/decoder.spec.ts +++ b/packages/node/src/solana/decoder.spec.ts @@ -11,9 +11,7 @@ import { SolanaDecoder } from './decoder'; import { getProgramId, filterInstructionsProcessor } from './utils.solana'; 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_codama_0_1_0: IdlV01 = require('../../test/8t2R21V3vjS1ucZzmX2memtGptjYZi2yGY3cYVa8dak7.idl.json'); const IDL_Jupiter: IdlV01 = require('../../test/JUP6LkbZbjS1jKKwapdHNy74zcZ3tLUZoi5QNyVTaV4.idl.json'); diff --git a/yarn.lock b/yarn.lock index c776eed48..0aa45545e 100644 --- a/yarn.lock +++ b/yarn.lock @@ -2872,6 +2872,20 @@ __metadata: languageName: node linkType: hard +"@solana/errors@npm:^3.0.2": + version: 3.0.3 + resolution: "@solana/errors@npm:3.0.3" + dependencies: + chalk: "npm:5.6.2" + commander: "npm:14.0.0" + peerDependencies: + typescript: ">=5.3.3" + bin: + errors: bin/cli.mjs + checksum: 10/33f08313dc3101d41aa4465f747ba4b4caec4db7745f3b651c7c436c2df1ec2d128b25ff676be548747916a5532bdefa859acde729f259b8f56de83b2ed43e05 + languageName: node + linkType: hard + "@solana/fast-stable-stringify@npm:3.0.2": version: 3.0.2 resolution: "@solana/fast-stable-stringify@npm:3.0.2" @@ -3378,6 +3392,7 @@ __metadata: "@nestjs/schedule": "npm:^5.0.1" "@nestjs/schematics": "npm:^11.0.0" "@nestjs/testing": "npm:^11.0.8" + "@solana/errors": "npm:^3.0.2" "@solana/kit": "npm:^3.0.2" "@subql/common-solana": "workspace:*" "@subql/node-core": "npm:^18.3.3" @@ -5209,6 +5224,13 @@ __metadata: languageName: node linkType: hard +"chalk@npm:5.6.2": + version: 5.6.2 + resolution: "chalk@npm:5.6.2" + checksum: 10/1b2f48f6fba1370670d5610f9cd54c391d6ede28f4b7062dd38244ea5768777af72e5be6b74fb6c6d54cb84c4a2dff3f3afa9b7cb5948f7f022cfd3d087989e0 + languageName: node + linkType: hard + "chalk@npm:^2.0.0, chalk@npm:^2.3.2": version: 2.4.2 resolution: "chalk@npm:2.4.2" @@ -5521,7 +5543,7 @@ __metadata: languageName: node linkType: hard -"commander@npm:^14.0.0": +"commander@npm:14.0.0, commander@npm:^14.0.0": version: 14.0.0 resolution: "commander@npm:14.0.0" checksum: 10/c05418bfc35a3e8b5c67bd9f75f5b773f386f9b85f83e70e7c926047f270929cb06cf13cd68f387dd6e7e23c6157de8171b28ba606abd3e6256028f1f789becf