validate fuel blocks consistency (#300)

common/changes/@subsquid/fuel-data/check-fuel-consistency_2024-06-20-09-49.json

@@ -0,0 +1,10 @@
+{
+  "changes": [
+    {
+      "packageName": "@subsquid/fuel-data",
+      "comment": "validate blocks consistency and retry query if invalid",
+      "type": "minor"
+    }
+  ],
+  "packageName": "@subsquid/fuel-data"
+}

common/changes/@subsquid/fuel-normalization/check-fuel-consistency_2024-06-20-09-49.json

@@ -0,0 +1,10 @@
+{
+  "changes": [
+    {
+      "packageName": "@subsquid/fuel-normalization",
+      "comment": "make block.transactions optional",
+      "type": "minor"
+    }
+  ],
+  "packageName": "@subsquid/fuel-normalization"
+}

fuel/fuel-data/src/data-source.ts

@@ -2,9 +2,10 @@ import {HttpClient} from '@subsquid/http-client'
 import {Batch, coldIngest} from '@subsquid/util-internal-ingest-tools'
 import {RangeRequest, SplitRequest} from '@subsquid/util-internal-range'
 import {DataValidationError, GetSrcType, Validator} from '@subsquid/util-internal-validation'
+import {wait} from '@subsquid/util-internal'
 import assert from 'assert'
 import {BlockData, Blocks, LatestBlockHeight, GetBlockHash, DataRequest, GetBlockHeader, BlockHeader} from './raw-data'
-import {getLatesBlockQuery, getBlockHashQuery, getBlockHeaderQuery, getBlocksQuery} from './query'
+import {getLatestBlockQuery, getBlockHashQuery, getBlockHeaderQuery, getBlocksQuery} from './query'
 
 
 function getResultValidator<V extends Validator>(validator: V): (result: unknown) => GetSrcType<V> {
@@ -42,7 +43,7 @@ export class HttpDataSource {
     }
 
     async getFinalizedHeight(): Promise<number> {
-        let query = getLatesBlockQuery()
+        let query = getLatestBlockQuery()
         let response: LatestBlockHeight = await this.request(query, getResultValidator(LatestBlockHeight))
         let height = parseInt(response.chain.latestBlock.header.height)
         assert(Number.isSafeInteger(height))
@@ -80,7 +81,16 @@ export class HttpDataSource {
         let first = req.range.to - req.range.from + 1
         let after = req.range.from == 0 ? undefined : req.range.from - 1
         let query = getBlocksQuery(req.request, first, after)
-        let response: Blocks = await this.request(query, getResultValidator(Blocks))
+
+        let response = await this.retry(async () => {
+            let response: Blocks = await this.request(query, getResultValidator(Blocks))
+            let length = response.blocks.nodes.length
+            if (length != first) {
+                throw new BlocksConsistencyError(`Expected blocks length: ${first}, got ${length}`)
+            }
+            return response
+        })
+
         let blocks = response.blocks.nodes.map(block => {
             let height = parseInt(block.header.height)
             assert(Number.isSafeInteger(height))
@@ -100,4 +110,28 @@ export class HttpDataSource {
         return this.client.graphqlRequest(query, {retryAttempts: Number.MAX_SAFE_INTEGER})
             .then(res => validateResult ? validateResult(res) : res)
     }
+
+    private async retry<T>(request: () => Promise<T>): Promise<T> {
+        let retries = 0
+        let pause = 200
+        while (true) {
+            try {
+                return await request()
+            } catch(err: any) {
+                if (err instanceof BlocksConsistencyError && retries < 3) {
+                    retries += 1
+                    await wait(pause)
+                } else {
+                    throw err
+                }
+            }
+        }
+    }
+}
+
+
+export class BlocksConsistencyError extends Error {
+    get name(): string {
+        return 'BlocksConsistencyError'
+    }
 }

fuel/fuel-data/src/query.ts

@@ -2,7 +2,7 @@ import {Output} from '@subsquid/util-internal-code-printer'
 import {DataRequest} from './raw-data';
 
 
-export function getLatesBlockQuery() {
+export function getLatestBlockQuery() {
     return `
         {
             chain {

fuel/fuel-data/src/raw-data.ts

@@ -303,7 +303,7 @@ export type Transaction = GetSrcType<typeof Transaction>
 
 export const Block = object({
     header: BlockHeader,
-    transactions: array(Transaction),
+    transactions: option(array(Transaction)),
 })
 
 

fuel/fuel-normalization/src/mapping.ts

@@ -392,7 +392,7 @@ export function mapRawBlock(raw: raw.BlockData): Block {
     let outputs: TransactionOutput[] = []
     let receipts: Receipt[] = []
 
-    raw.block.transactions.forEach((tx, index) => {
+    raw.block.transactions?.forEach((tx, index) => {
         transactions.push(mapRawTransaction(tx, index))
 
         tx.inputs?.forEach((input, inputIndex) => {