Implement Effective Reorg Handling Mechanism

[aruokhai]

Objective
Reorgs Can be fatal to a blockchain indexer if not handled properly, this is because the during Reorgs, blocks are replaced with other blocks, leading to inconsistent state of the indexer. This inconsistency must be handled effectively to ensure correctness of such indexer.

Changes

• Implemented Reorg Mechanism
• Improved OperationState handling logic
• Added Dynamic Cron Job Scheduling Mechanism
• Added Verbosity Logging Option.

Scope Of Change
-The scope of change is critical because it affects the core functioning of the indexer.

[theanmolsharma]

Concept ACK
I like the approach although I don't love it.
I don't like the fact that we have to store the entire hash chain. I'd look at other RPCs like getbestblockhash or something.

More in-depth review coming soon.

[aruokhai]

Concept ACK I like the approach although I don't love it. I don't like the fact that we have to store the entire hash chain. I'd look at other RPCs like getbestblockhash or something.

More in-depth review coming soon.

We Only store 6 blockhash or whatever number is necessary, if reorg is levels deep, we result to error.

[theanmolsharma]

Concept ACK I like the approach although I don't love it. I don't like the fact that we have to store the entire hash chain. I'd look at other RPCs like getbestblockhash or something.
More in-depth review coming soon.

We Only store 6 blockhash or whatever number is necessary, if reorg is levels deep, we result to error.

This is not good, ideally we should be able to handle reorgs of any length. Reorgs greater than 6 are rare, but we should be prepared for the worst. Let's store the last 2016 blocks to be very conservative.

[theanmolsharma]

nice work!!

[theanmolsharma]

Here's a diff that addresses most but not all comments

diff --git a/src/block-data-providers/base-block-data-provider.abstract.ts b/src/block-data-providers/base-block-data-provider.abstract.ts
index 05b5e6c..e631d69 100644
--- a/src/block-data-providers/base-block-data-provider.abstract.ts
+++ b/src/block-data-providers/base-block-data-provider.abstract.ts
@@ -7,6 +7,7 @@ import {
 } from '@/indexer/indexer.service';
 import { ConfigService } from '@nestjs/config';
 import { BlockStateService } from '@/block-state/block-state.service';
+import { BlockState } from '@/block-state/block-state.entity';
 
 export abstract class BaseBlockDataProvider<OperationState> {
     protected abstract readonly logger: Logger;
@@ -43,11 +44,16 @@ export abstract class BaseBlockDataProvider<OperationState> {
         )?.state;
     }
 
-    async setState(state: OperationState): Promise<void> {
+    async setState(
+        state: OperationState,
+        blockState: BlockState,
+    ): Promise<void> {
         await this.operationStateService.setOperationState(
             this.operationStateKey,
             state,
         );
+
+        await this.blockStateService.addBlockState(blockState);
     }
 
     abstract getBlockHash(height: number): Promise<string>;
@@ -55,24 +61,16 @@ export abstract class BaseBlockDataProvider<OperationState> {
     async traceReorg(): Promise<number> {
         let state = await this.blockStateService.getCurrentBlockState();
 
-        if (state == null) {
-            return null;
-        }
+        if (state === null) return null;
 
         while (true) {
-            if (state === null) {
-                throw new Error('Reorgs levels deep');
-            }
+            if (state === null) throw new Error('Reorgs levels deep');
 
-            const fetchedBlockHash = await this.getBlockHash(
-                state.indexedBlockHeight,
-            );
+            const fetchedBlockHash = await this.getBlockHash(state.blockHeight);
 
-            if (state.indexedBlockHash === fetchedBlockHash) {
-                return state.indexedBlockHeight;
-            }
+            if (state.blockHash === fetchedBlockHash) return state.blockHeight;
 
-            state = await this.blockStateService.dequeue_operation_state();
+            state = await this.blockStateService.dequeueState();
         }
     }
 }
diff --git a/src/block-data-providers/bitcoin-core/provider.ts b/src/block-data-providers/bitcoin-core/provider.ts
index 5c8b9a4..b44bd16 100644
--- a/src/block-data-providers/bitcoin-core/provider.ts
+++ b/src/block-data-providers/bitcoin-core/provider.ts
@@ -73,15 +73,23 @@ export class BitcoinCoreProvider
             );
         } else {
             this.logger.log('No previous state found. Starting from scratch.');
-            const updatedState: BitcoinCoreOperationState = {
-                indexedBlockHeight:
-                    this.configService.get<BitcoinNetwork>('app.network') ===
-                    BitcoinNetwork.MAINNET
-                        ? TAPROOT_ACTIVATION_HEIGHT - 1
-                        : 0,
-            };
-
-            await this.setState(updatedState);
+
+            const blockHeight =
+                this.configService.get<BitcoinNetwork>('app.network') ===
+                BitcoinNetwork.MAINNET
+                    ? TAPROOT_ACTIVATION_HEIGHT - 1
+                    : 0;
+            const blockHash = await this.getBlockHash(blockHeight);
+
+            await this.setState(
+                {
+                    indexedBlockHeight: blockHeight,
+                },
+                {
+                    blockHash,
+                    blockHeight,
+                },
+            );
         }
     }
 
@@ -132,11 +140,9 @@ export class BitcoinCoreProvider
                 }
 
                 state.indexedBlockHeight = height;
-                await this.setState(state);
-
-                await this.blockStateService.addBlockState({
-                    indexedBlockHash: blockHash,
-                    indexedBlockHeight: height,
+                await this.setState(state, {
+                    blockHash: blockHash,
+                    blockHeight: height,
                 });
             }
         } finally {
diff --git a/src/block-data-providers/esplora/provider.ts b/src/block-data-providers/esplora/provider.ts
index 0403979..5dc36f5 100644
--- a/src/block-data-providers/esplora/provider.ts
+++ b/src/block-data-providers/esplora/provider.ts
@@ -71,16 +71,25 @@ export class EsploraProvider
             );
         } else {
             this.logger.log('No previous state found. Starting from scratch.');
-            const updatedState: EsploraOperationState = {
-                currentBlockHeight: 0,
-                indexedBlockHeight:
-                    this.configService.get<BitcoinNetwork>('app.network') ===
-                    BitcoinNetwork.MAINNET
-                        ? TAPROOT_ACTIVATION_HEIGHT - 1
-                        : 0,
-                lastProcessedTxIndex: 0, // we dont take coinbase txn in account
-            };
-            await this.setState(updatedState);
+
+            const blockHeight =
+                this.configService.get<BitcoinNetwork>('app.network') ===
+                BitcoinNetwork.MAINNET
+                    ? TAPROOT_ACTIVATION_HEIGHT - 1
+                    : 0;
+            const blockHash = await this.getBlockHash(blockHeight);
+
+            await this.setState(
+                {
+                    currentBlockHeight: 0,
+                    indexedBlockHeight: blockHeight,
+                    lastProcessedTxIndex: 0, // we don't take coinbase txn into account
+                },
+                {
+                    blockHash,
+                    blockHeight,
+                },
+            );
         }
     }
 
@@ -162,11 +171,9 @@ export class EsploraProvider
 
                 state.indexedBlockHeight = height;
                 state.lastProcessedTxIndex = i + this.batchSize - 1;
-                await this.setState(state);
-
-                await this.blockStateService.addBlockState({
-                    indexedBlockHash: hash,
-                    indexedBlockHeight: height,
+                await this.setState(state, {
+                    blockHeight: height,
+                    blockHash: hash,
                 });
             } catch (error) {
                 this.logger.error(
diff --git a/src/block-state/block-state.entity.ts b/src/block-state/block-state.entity.ts
index 661700f..f31a7d7 100644
--- a/src/block-state/block-state.entity.ts
+++ b/src/block-state/block-state.entity.ts
@@ -3,8 +3,8 @@ import { Column, Entity, PrimaryColumn } from 'typeorm';
 @Entity()
 export class BlockState {
     @PrimaryColumn('integer')
-    indexedBlockHeight: number;
+    blockHeight: number;
 
     @Column('text')
-    indexedBlockHash: string;
+    blockHash: string;
 }
diff --git a/src/block-state/block-state.service.ts b/src/block-state/block-state.service.ts
index e1e53b8..0faa6d9 100644
--- a/src/block-state/block-state.service.ts
+++ b/src/block-state/block-state.service.ts
@@ -15,60 +15,32 @@ export class BlockStateService {
     ) {}
 
     async getCurrentBlockState(): Promise<BlockState> {
-        const state = (
-            await this.blockStateRepository.find({
-                order: {
-                    indexedBlockHeight: 'DESC',
-                },
-                take: 1,
-            })
-        )[0];
-
-        return state;
+        return this.blockStateRepository.findOne({
+            order: {
+                blockHeight: 'DESC',
+            },
+        });
     }
 
     async addBlockState(state: BlockState): Promise<void> {
-        this.blockStateRepository.save(state);
-        // Ensure the cache size does not exceed 2016
-        await this.trimState();
+        await this.blockStateRepository.save(state);
     }
 
     // Remove and return the latest item in the state cache
-    async dequeue_operation_state(): Promise<BlockState | null> {
-        const latest_state = (
-            await this.blockStateRepository.find({
-                order: {
-                    indexedBlockHeight: 'DESC',
-                },
-                take: 1,
-            })
-        )[0];
-
-        if (latest_state) {
-            await this.blockStateRepository.remove(latest_state);
+    async dequeueState(): Promise<BlockState | null> {
+        const latestState = await this.blockStateRepository.findOne({
+            order: {
+                blockHeight: 'DESC',
+            },
+        });
+
+        if (latestState) {
+            await this.blockStateRepository.remove(latestState);
             await this.transactionService.deleteTransactionByBlockHash(
-                latest_state.indexedBlockHash,
+                latestState.blockHash,
             );
-            return latest_state;
         }
 
-        return null;
-    }
-
-    private async trimState(): Promise<void> {
-        const queueCount = await this.blockStateRepository.count();
-
-        if (queueCount >= this.cacheSize) {
-            // Delete the oldest entries from the cache
-            const old_states = await this.blockStateRepository.find({
-                order: {
-                    indexedBlockHeight: 'ASC',
-                },
-                take: queueCount - this.cacheSize + 1,
-            });
-            await this.blockStateRepository.delete(
-                old_states.map((state) => state.indexedBlockHeight),
-            );
-        }
+        return latestState;
     }
 }
diff --git a/src/main.ts b/src/main.ts
index 68f5ccf..6488ce1 100644
--- a/src/main.ts
+++ b/src/main.ts
@@ -12,18 +12,12 @@ async function bootstrap() {
     const port = configService.get<number>('app.port');
 
     const isVerbose = configService.get<boolean>('app.verbose') ?? false;
-
     const isDebug = configService.get<boolean>('app.debug') ?? false;
 
     const loggerLevels: LogLevel[] = ['error', 'warn', 'log'];
 
-    if (isVerbose) {
-        loggerLevels.push('verbose');
-    }
-
-    if (isDebug) {
-        loggerLevels.push('debug');
-    }
+    if (isVerbose) loggerLevels.push('verbose');
+    if (isDebug) loggerLevels.push('debug');
 
     app.useLogger(loggerLevels);
 

[theanmolsharma]

Consider approved!! I need to run some reorg simulations to verify reorg handling. Will merge, once I'm confident that it works.

[theanmolsharma]

ACK 2327412

Excellent work!