Implement Effective Reorg Handling Mechanism

.gitignore

@@ -29,7 +29,6 @@ lerna-debug.log*
 
 # IDE - VSCode
 .vscode/*
-!.vscode/settings.json
 !.vscode/tasks.json
 !.vscode/launch.json
 !.vscode/extensions.json

config/config.yaml

@@ -3,6 +3,8 @@ db:
   synchronize: false
 app:
   port: <port>
+  verbose: <boolean>
+  debug: <boolean>
   network: <network>
   requestRetry:
     delay: <number> # delay in Milliseconds

config/dev.config.yaml

@@ -3,18 +3,19 @@ db:
   synchronize: true
 app:
   port: 3000
+  verbose: false
+  debug: true
   network: regtest
   requestRetry:
     delay: 3000
     count: 3
-providerType: ESPLORA
+providerType: BITCOIN_CORE_RPC
 esplora:
   url: https://blockstream.info
   batchSize: 5
 bitcoinCore:
   protocol: http
   rpcHost: 127.0.0.1
-  rpcPass: polarpass
-  rpcUser: polaruser
-  rpcPort: 18445
-
+  rpcPass: password
+  rpcUser: admin
+  rpcPort: 18443

config/e2e.config.yaml

@@ -3,6 +3,8 @@ db:
   synchronize: true
 app:
   port: 3000
+  verbose: false
+  debug: true
   network: regtest
   requestRetry:
     delay: 500

src/block-data-providers/base-block-data-provider.abstract.ts

@@ -5,14 +5,19 @@ import {
     TransactionInput,
     TransactionOutput,
 } 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;
     protected abstract readonly operationStateKey: string;
 
     protected constructor(
+        protected readonly configService: ConfigService,
         private readonly indexerService: IndexerService,
         private readonly operationStateService: OperationStateService,
+        protected readonly blockStateService: BlockStateService,
     ) {}
 
     async indexTransaction(
@@ -39,10 +44,38 @@ 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>;
+
+    async traceReorg(): Promise<number> {
+        let state = await this.blockStateService.getCurrentBlockState();
+
+        if (!state) return null;
+
+        while (state) {
+            const fetchedBlockHash = await this.getBlockHash(state.blockHeight);
+
+            if (state.blockHash === fetchedBlockHash) return state.blockHeight;
+
+            await this.blockStateService.removeState(state);
+
+            this.logger.log(
+                `Reorg found at height: ${state.blockHeight}, Wrong hash: ${state.blockHash}, Correct hash: ${fetchedBlockHash}`,
+            );
+            state = await this.blockStateService.getCurrentBlockState();
+        }
+
+        throw new Error('Cannot Reorgs, blockchain state exhausted');
     }
 }

src/block-data-providers/bitcoin-core/interfaces.ts

@@ -40,7 +40,6 @@ export interface Output {
 }
 
 export type BitcoinCoreOperationState = {
-    currentBlockHeight: number;
     indexedBlockHeight: number;
 };
 

src/block-data-providers/bitcoin-core/provider.spec.ts

@@ -12,6 +12,7 @@ import {
     rawTransactions,
 } from '@/block-data-providers/bitcoin-core/provider-fixtures';
 import { Test, TestingModule } from '@nestjs/testing';
+import { BlockStateService } from '@/block-state/block-state.service';
 
 describe('Bitcoin Core Provider', () => {
     let provider: BitcoinCoreProvider;
@@ -46,6 +47,10 @@ describe('Bitcoin Core Provider', () => {
                         getOperationState: jest.fn(),
                     },
                 },
+                {
+                    provide: BlockStateService,
+                    useClass: jest.fn(),
+                },
             ],
         }).compile();
 
@@ -72,8 +77,8 @@ describe('Bitcoin Core Provider', () => {
 
     it('should process each transaction of a block appropriately', async () => {
         const result = await provider.processBlock(3, 2);
-        expect(result).toHaveLength(1);
-        expect(result).toEqual(
+        expect(result[0]).toHaveLength(1);
+        expect(result[0]).toEqual(
             expect.arrayContaining([...parsedTransactions.values()]),
         );
     });

src/block-data-providers/bitcoin-core/provider.ts

@@ -28,6 +28,7 @@ import {
 import { AxiosRequestConfig } from 'axios';
 import * as currency from 'currency.js';
 import { AxiosRetryConfig, makeRequest } from '@/common/request';
+import { BlockStateService } from '@/block-state/block-state.service';
 
 @Injectable()
 export class BitcoinCoreProvider
@@ -41,11 +42,17 @@ export class BitcoinCoreProvider
     private retryConfig: AxiosRetryConfig;
 
     public constructor(
-        private readonly configService: ConfigService,
+        configService: ConfigService,
         indexerService: IndexerService,
         operationStateService: OperationStateService,
+        blockStateService: BlockStateService,
     ) {
-        super(indexerService, operationStateService);
+        super(
+            configService,
+            indexerService,
+            operationStateService,
+            blockStateService,
+        );
 
         const { protocol, rpcPort, rpcHost } =
             configService.get<BitcoinCoreConfig>('bitcoinCore');
@@ -57,25 +64,32 @@ export class BitcoinCoreProvider
     }
 
     async onApplicationBootstrap() {
-        const getState = await this.getState();
-        if (getState) {
+        const currentState = await this.getState();
+        if (currentState) {
             this.logger.log(
                 `Restoring state from previous run: ${JSON.stringify(
-                    getState,
+                    currentState,
                 )}`,
             );
         } else {
             this.logger.log('No previous state found. Starting from scratch.');
-            const state: BitcoinCoreOperationState = {
-                currentBlockHeight: 0,
-                indexedBlockHeight:
-                    this.configService.get<BitcoinNetwork>('app.network') ===
-                    BitcoinNetwork.MAINNET
-                        ? TAPROOT_ACTIVATION_HEIGHT - 1
-                        : 0,
-            };
-
-            await this.setState(state);
+
+            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,
+                },
+            );
         }
     }
 
@@ -85,12 +99,14 @@ export class BitcoinCoreProvider
         this.isSyncing = true;
 
         const state = await this.getState();
+
         if (!state) {
             throw new Error('State not found');
         }
 
         try {
             const tipHeight = await this.getTipHeight();
+
             if (tipHeight <= state.indexedBlockHeight) {
                 this.logger.debug(
                     `No new blocks found. Current tip height: ${tipHeight}`,
@@ -102,9 +118,11 @@ export class BitcoinCoreProvider
             const networkInfo = await this.getNetworkInfo();
             const verbosityLevel = this.versionToVerbosity(networkInfo.version);
 
-            let height = state.indexedBlockHeight + 1;
+            let height =
+                ((await this.traceReorg()) ?? state.indexedBlockHeight) + 1;
+
             for (height; height <= tipHeight; height++) {
-                const transactions = await this.processBlock(
+                const [transactions, blockHash] = await this.processBlock(
                     height,
                     verbosityLevel,
                 );
@@ -122,7 +140,10 @@ export class BitcoinCoreProvider
                 }
 
                 state.indexedBlockHeight = height;
-                await this.setState(state);
+                await this.setState(state, {
+                    blockHash: blockHash,
+                    blockHeight: height,
+                });
             }
         } finally {
             this.isSyncing = false;
@@ -143,7 +164,7 @@ export class BitcoinCoreProvider
         });
     }
 
-    private async getBlockHash(height: number): Promise<string> {
+    async getBlockHash(height: number): Promise<string> {
         return this.request({
             method: 'getblockhash',
             params: [height],
@@ -170,7 +191,7 @@ export class BitcoinCoreProvider
     public async processBlock(
         height: number,
         verbosityLevel: number,
-    ): Promise<Transaction[]> {
+    ): Promise<[Transaction[], string]> {
         const parsedTransactionList: Transaction[] = [];
         const blockHash = await this.getBlockHash(height);
         this.logger.debug(
@@ -188,7 +209,7 @@ export class BitcoinCoreProvider
             parsedTransactionList.push(parsedTransaction);
         }
 
-        return parsedTransactionList;
+        return [parsedTransactionList, blockHash];
     }
 
     private async parseTransaction(

src/block-data-providers/block-provider.module.ts

@@ -7,31 +7,46 @@ import { IndexerService } from '@/indexer/indexer.service';
 import { ProviderType } from '@/common/enum';
 import { BitcoinCoreProvider } from '@/block-data-providers/bitcoin-core/provider';
 import { EsploraProvider } from '@/block-data-providers/esplora/provider';
+import { BlockStateService } from '@/block-state/block-state.service';
+import { BlockStateModule } from '@/block-state/block-state.module';
 
 @Module({
-    imports: [OperationStateModule, IndexerModule, ConfigModule],
+    imports: [
+        OperationStateModule,
+        IndexerModule,
+        ConfigModule,
+        BlockStateModule,
+    ],
     controllers: [],
     providers: [
         {
             provide: 'BlockDataProvider',
-            inject: [ConfigService, IndexerService, OperationStateService],
+            inject: [
+                ConfigService,
+                IndexerService,
+                OperationStateService,
+                BlockStateService,
+            ],
             useFactory: (
                 configService: ConfigService,
                 indexerService: IndexerService,
                 operationStateService: OperationStateService,
+                blockStateService: BlockStateService,
             ) => {
                 switch (configService.get<ProviderType>('providerType')) {
                     case ProviderType.ESPLORA:
                         return new EsploraProvider(
                             configService,
                             indexerService,
                             operationStateService,
+                            blockStateService,
                         );
                     case ProviderType.BITCOIN_CORE_RPC:
                         return new BitcoinCoreProvider(
                             configService,
                             indexerService,
                             operationStateService,
+                            blockStateService,
                         );
                     default:
                         throw Error('unrecognised provider type in config');

src/block-data-providers/esplora/interface.ts

@@ -1,8 +1,8 @@
-export type EsploraOperationState = {
+export interface EsploraOperationState {
     currentBlockHeight: number;
     indexedBlockHeight: number;
     lastProcessedTxIndex: number;
-};
+}
 
 type EsploraTransactionInput = {
     txid: string;