substrate: allow to configure finalityConfirmation in substrate proce…

…ssor (#315)

* add finalityConfirmation to substrate processor

* simplify getFinalizedHeight if finalityConfirmation is specified

* retry in case of reorg

* retry both rpc calls

common/changes/@subsquid/substrate-data-raw/substrate-finality_2024-07-22-00-43.json

@@ -0,0 +1,10 @@
+{
+  "changes": [
+    {
+      "packageName": "@subsquid/substrate-data-raw",
+      "comment": "allow to use finalityConfirmation to determine finalized blocks",
+      "type": "minor"
+    }
+  ],
+  "packageName": "@subsquid/substrate-data-raw"
+}

common/changes/@subsquid/substrate-data/substrate-finality_2024-07-22-00-43.json

@@ -0,0 +1,10 @@
+{
+  "changes": [
+    {
+      "packageName": "@subsquid/substrate-data",
+      "comment": "accept finalityConfirmation in `RpcDataSourceOptions`",
+      "type": "minor"
+    }
+  ],
+  "packageName": "@subsquid/substrate-data"
+}

common/changes/@subsquid/substrate-processor/substrate-finality_2024-07-22-00-43.json

@@ -0,0 +1,10 @@
+{
+  "changes": [
+    {
+      "packageName": "@subsquid/substrate-processor",
+      "comment": "allow to use finalityConfirmation if `chain_getFinalizedHead` is not an option",
+      "type": "minor"
+    }
+  ],
+  "packageName": "@subsquid/substrate-processor"
+}

substrate/substrate-data-raw/src/datasource.ts

@@ -1,13 +1,12 @@
 import {Logger, LogLevel} from '@subsquid/logger'
-import {RpcClient} from '@subsquid/rpc-client'
+import {RpcClient, SubscriptionHandle} from '@subsquid/rpc-client'
 import {AsyncQueue, ensureError, maybeLast, partitionBy, Throttler, wait} from '@subsquid/util-internal'
 import {
     assertIsValid,
     Batch,
     BlockConsistencyError,
     BlockRef,
     ChainHeads,
-    DataConsistencyError,
     HashAndHeight,
     HotProcessor,
     HotState,
@@ -34,6 +33,7 @@ export interface RpcDataSourceOptions {
     rpc: RpcClient
     headPollInterval?: number
     newHeadTimeout?: number
+    finalityConfirmation?: number
     log?: Logger
 }
 
@@ -42,20 +42,28 @@ export class RpcDataSource {
     public readonly rpc: Rpc
     private headPollInterval: number
     private newHeadTimeout: number
+    private finalityConfirmation?: number
     private log?: Logger
 
     constructor(options: RpcDataSourceOptions) {
         this.rpc = new Rpc(options.rpc)
         this.headPollInterval = options.headPollInterval ?? 5000
         this.newHeadTimeout = options.newHeadTimeout ?? 0
+        this.finalityConfirmation = options.finalityConfirmation
         this.log = options.log
     }
 
     async getFinalizedHeight(): Promise<number> {
-        let head = await this.rpc.getFinalizedHead()
-        let header = await this.rpc.getBlockHeader(head)
-        assert(header, 'finalized blocks must be always available')
-        return qty2Int(header.number)
+        if (this.finalityConfirmation == null) {
+            let head = await this.rpc.getFinalizedHead()
+            let header = await this.rpc.getBlockHeader(head)
+            assert(header, 'finalized blocks must be always available')
+            return qty2Int(header.number)
+        } else {
+            let header = await this.rpc.getBlockHeader()
+            assert(header, 'the header of the latest block on the chain must be always available')
+            return Math.max(0, qty2Int(header.number) - this.finalityConfirmation)
+        }
     }
 
     async *getFinalizedBlocks(requests: RangeRequestList<DataRequest>, stopOnHead?: boolean): AsyncIterable<Batch<BlockData>> {
@@ -129,7 +137,7 @@ export class RpcDataSource {
                 for (let split of splitRangeByRequest(requests, {from, to: top})) {
                     for (let range of splitRange(10, split.range)) {
                         let blocks = await fetch.getHotSplit(
-                            from,
+                            range.from,
                             range.to === headBlock?.height ? headBlock : range.to,
                             split.request || {}
                         )
@@ -172,7 +180,25 @@ export class RpcDataSource {
         while (!isEnd()) {
             let head = await headSrc.call()
             if (head === prev) continue
-            let finalizedHead = await this.rpc.getFinalizedHead()
+
+            let finalizedHead: string | null = null
+            if (this.finalityConfirmation == null) {
+                finalizedHead = await this.rpc.getFinalizedHead()
+            } else {
+                let attempt = 0
+                while (attempt < 5) {
+                    let header = await this.rpc.getBlockHeader(head)
+                    if (header != null) {
+                        let finalizedHeight = qty2Int(header.number) - this.finalityConfirmation
+                        finalizedHead = await this.rpc.getBlockHash(finalizedHeight)
+                        if (finalizedHead != null) break
+                    }
+
+                    head = await this.rpc.getHead()
+                    attempt += 1
+                }
+                assert(finalizedHead != null, `failed to get finalized head after ${attempt} attempts`)
+            }
             await this.handleNewHeads({
                 best: {hash: head},
                 finalized: {hash: finalizedHead}
@@ -181,27 +207,31 @@ export class RpcDataSource {
     }
 
     private async subscription(cb: (heads: ChainHeads) => Promise<void>, isEnd: () => boolean): Promise<void> {
+        let finalityConfirmation = this.finalityConfirmation
         let queue = new AsyncQueue<number | Error>(1)
         let finalizedHeight = 0
         let prevHeight = 0
 
-        let finalizedHeadsHandle = this.rpc.client.subscribe<BlockHeader>({
-            method: 'chain_subscribeFinalizedHeads',
-            unsubscribe: 'chain_unsubscribeFinalizedHeads',
-            notification: 'chain_finalizedHead',
-            onMessage(head: BlockHeader) {
-                try {
-                    let height = qty2Int(head.number)
-                    finalizedHeight = Math.max(finalizedHeight, height)
-                } catch(err: any) {
-                    close(err)
-                }
-            },
-            onError(err: Error) {
-                close(ensureError(err))
-            },
-            resubscribeOnConnectionLoss: true
-        })
+        let finalizedHeadsHandle: SubscriptionHandle | undefined
+        if (finalityConfirmation == null) {
+            finalizedHeadsHandle = this.rpc.client.subscribe<BlockHeader>({
+                method: 'chain_subscribeFinalizedHeads',
+                unsubscribe: 'chain_unsubscribeFinalizedHeads',
+                notification: 'chain_finalizedHead',
+                onMessage(head: BlockHeader) {
+                    try {
+                        let height = qty2Int(head.number)
+                        finalizedHeight = Math.max(finalizedHeight, height)
+                    } catch(err: any) {
+                        close(err)
+                    }
+                },
+                onError(err: Error) {
+                    close(ensureError(err))
+                },
+                resubscribeOnConnectionLoss: true
+            })
+        }
 
         let newHeadsHandle = this.rpc.client.subscribe<BlockHeader>({
             method: 'chain_subscribeNewHeads',
@@ -212,6 +242,9 @@ export class RpcDataSource {
                     let height = qty2Int(head.number)
                     if (height >= prevHeight) {
                         prevHeight = height
+                        if (finalityConfirmation != null) {
+                            finalizedHeight = Math.max(0, height - finalityConfirmation)
+                        }
                         queue.forcePut(height)
                     }
                 } catch(err: any) {
@@ -226,7 +259,7 @@ export class RpcDataSource {
 
         function close(err?: Error) {
             newHeadsHandle.close()
-            finalizedHeadsHandle.close()
+            finalizedHeadsHandle?.close()
             if (err) {
                 queue.forcePut(err)
             }

substrate/substrate-data-raw/src/rpc.ts

@@ -37,7 +37,7 @@ export class Rpc {
         return this.call('chain_getBlockHash', [toQty(height)])
     }
 
-    getBlockHeader(blockHash: Hash): Promise<BlockHeader | null> {
+    getBlockHeader(blockHash?: Hash): Promise<BlockHeader | null> {
         return this.call('chain_getHeader', [blockHash])
     }
 

substrate/substrate-data/src/datasource.ts

@@ -12,6 +12,7 @@ export interface RpcDataSourceOptions {
     headPollInterval?: number
     newHeadTimeout?: number
     typesBundle?: OldTypesBundle | OldSpecsBundle
+    finalityConfirmation?: number
 }
 
 
@@ -23,7 +24,8 @@ export class RpcDataSource {
         this.rawDataSource = new raw.RpcDataSource({
             rpc: options.rpc,
             headPollInterval: options.headPollInterval,
-            newHeadTimeout: options.newHeadTimeout
+            newHeadTimeout: options.newHeadTimeout,
+            finalityConfirmation: options.finalityConfirmation
         })
         this.typesBundle = options.typesBundle
     }

substrate/substrate-processor/src/ds-rpc.ts

@@ -15,6 +15,7 @@ export interface RpcDataSourceOptions {
     headPollInterval?: number
     newHeadTimeout?: number
     typesBundle?: OldTypesBundle | OldSpecsBundle
+    finalityConfirmation?: number
 }
 
 

substrate/substrate-processor/src/processor.ts

@@ -159,6 +159,7 @@ export class SubstrateBatchProcessor<F extends FieldSelection = {}> {
     private requests: RangeRequest<DataRequest>[] = []
     private fields?: FieldSelection
     private blockRange?: Range
+    private finalityConfirmation?: number
     private archive?: GatewaySettings
     private rpcEndpoint?: RpcEndpointSettings
     private rpcIngestSettings?: RpcDataIngestionSettings
@@ -285,6 +286,18 @@ export class SubstrateBatchProcessor<F extends FieldSelection = {}> {
         return this
     }
 
+    /**
+     * Distance from the head block behind which all blocks are considered to be finalized.
+     *
+     * By default, the processor will track finalized blocks via `chain_getFinalizedHead`.
+     * Configure it only if `chain_getFinalizedHead` doesn’t return the expected info.
+     */
+    setFinalityConfirmation(nBlocks: number): this {
+        this.assertNotRunning()
+        this.finalityConfirmation = nBlocks
+        return this
+    }
+
     /**
      * Configure a set of fetched fields
      */
@@ -456,7 +469,8 @@ export class SubstrateBatchProcessor<F extends FieldSelection = {}> {
             rpc: this.getChainRpcClient(),
             headPollInterval: this.rpcIngestSettings?.headPollInterval,
             newHeadTimeout: this.rpcIngestSettings?.newHeadTimeout,
-            typesBundle: this.typesBundle
+            typesBundle: this.typesBundle,
+            finalityConfirmation: this.finalityConfirmation
         })
     }
 
@@ -586,6 +600,7 @@ export class SubstrateBatchProcessor<F extends FieldSelection = {}> {
                 requests: this.getBatchRequests(),
                 archive: this.archive == null ? undefined : this.getArchiveDataSource(),
                 hotDataSource: this.rpcIngestSettings?.disabled ? undefined : this.getRpcDataSource(),
+                allBlocksAreFinal: this.finalityConfirmation === 0,
                 process: (s, b) => this.processBatch(s, b as any, handler),
                 prometheus: this.prometheus,
                 log