11/* eslint-disable no-underscore-dangle */
22/* eslint-disable no-await-in-loop */
33import '@frequency-chain/api-augment' ;
4- import { Injectable , Logger , OnApplicationBootstrap } from '@nestjs/common' ;
4+ import { Injectable , Logger , OnApplicationBootstrap , OnApplicationShutdown } from '@nestjs/common' ;
55import { InjectQueue } from '@nestjs/bullmq' ;
66import Redis from 'ioredis' ;
77import { InjectRedis } from '@songkeys/nestjs-redis' ;
@@ -15,14 +15,17 @@ import { EVENTS_TO_WATCH_KEY, LAST_SEEN_BLOCK_NUMBER_SCANNER_KEY, REGISTERED_WEB
1515import { ChainWatchOptionsDto } from '../dtos/chain.watch.dto' ;
1616import * as RedisUtils from '../utils/redis' ;
1717import { ChainEventProcessorService } from '../blockchain/chain-event-processor.service' ;
18+ import { IScanReset } from '../interfaces/scan-reset.interface' ;
19+
20+ const INTERVAL_SCAN_NAME = 'intervalScan' ;
1821
1922@Injectable ( )
20- export class ScannerService implements OnApplicationBootstrap {
23+ export class ScannerService implements OnApplicationBootstrap , OnApplicationShutdown {
2124 private readonly logger : Logger ;
2225
2326 private scanInProgress = false ;
24-
2527 private paused = false ;
28+ private scanResetBlockNumber : number | undefined ;
2629
2730 constructor (
2831 private readonly configService : ConfigService ,
@@ -36,32 +39,43 @@ export class ScannerService implements OnApplicationBootstrap {
3639 }
3740
3841 async onApplicationBootstrap ( ) {
39- const startingBlock = Number ( this . configService . startingBlock ) - 1 ;
40- this . setLastSeenBlockNumber ( startingBlock ) ;
41- this . scheduleInitialScan ( ) ;
42- this . scheduleBlockchainScan ( ) ;
43- }
44-
45- private scheduleInitialScan ( ) {
46- const initialTimeout = setTimeout ( ( ) => this . scan ( ) , 0 ) ;
47- this . schedulerRegistry . addTimeout ( 'initialScan' , initialTimeout ) ;
48- }
42+ const startingBlock = this . configService . startingBlock ;
43+ if ( startingBlock ) {
44+ this . logger . log ( `Setting initial scan block to ${ startingBlock } ) ;
45+ this . setLastSeenBlockNumber ( startingBlock - 1 ) ;
46+ }
47+ setImmediate ( ( ) => this . scan ( ) ) ;
4948
50- private scheduleBlockchainScan ( ) {
5149 const scanInterval = this . configService . blockchainScanIntervalMinutes * SECONDS_PER_MINUTE * MILLISECONDS_PER_SECOND ;
50+ this . schedulerRegistry . addInterval ( INTERVAL_SCAN_NAME , setInterval ( ( ) => this . scan ( ) , scanInterval ) ) ;
51+ }
5252
53- const interval = setInterval ( ( ) => this . scan ( ) , scanInterval ) ;
54- this . schedulerRegistry . addInterval ( 'blockchainScan' , interval ) ;
53+ onApplicationShutdown ( _signal ?: string | undefined ) {
54+ const interval = this . schedulerRegistry . getInterval ( INTERVAL_SCAN_NAME ) ;
55+ clearInterval ( interval ) ;
5556 }
5657
57- public async pauseScanner ( ) {
58+ public pauseScanner ( ) {
5859 this . logger . debug ( 'Pausing scanner' ) ;
5960 this . paused = true ;
6061 }
6162
62- public async resumeScanner ( ) {
63+ public resumeScanner ( immediate = false ) {
6364 this . logger . debug ( 'Resuming scanner' ) ;
6465 this . paused = false ;
66+ if ( immediate ) {
67+ setImmediate ( ( ) => this . scan ( ) ) ;
68+ }
69+ }
70+
71+ public async resetScan ( { blockNumber, rewindOffset, immediate } : IScanReset ) {
72+ this . pauseScanner ( ) ;
73+ let targetBlock = blockNumber ?? await this . blockchainService . getLatestFinalizedBlockNumber ( ) ;
74+ targetBlock -= rewindOffset ? Math . abs ( rewindOffset ) : 0 ;
75+ targetBlock = Math . max ( targetBlock , 1 ) ;
76+ this . scanResetBlockNumber = targetBlock ;
77+ this . logger . log ( `Resetting scan to block #${ targetBlock } ) ;
78+ this . resumeScanner ( immediate ) ;
6579 }
6680
6781 async scan ( ) {
@@ -73,18 +87,7 @@ export class ScannerService implements OnApplicationBootstrap {
7387 return ;
7488 }
7589
76- if ( this . paused ) {
77- this . logger . debug ( 'Scanner is paused' ) ;
78- return ;
79- }
80- let queueSize = await this . ipfsQueue . count ( ) ;
81-
82- if ( queueSize > 0 ) {
83- this . logger . log ( 'Deferring next blockchain scan until queue is empty' ) ;
84- return ;
85- }
8690 const registeredWebhook = await this . cache . get ( REGISTERED_WEBHOOK_KEY ) ;
87-
8891 if ( ! registeredWebhook ) {
8992 this . logger . log ( 'No registered webhooks; no scan performed.' ) ;
9093 return ;
@@ -93,32 +96,38 @@ export class ScannerService implements OnApplicationBootstrap {
9396 const eventsToWatch : ChainWatchOptionsDto = chainWatchFilters ? JSON . parse ( chainWatchFilters ) : { msa_ids : [ ] , schemaIds : [ ] } ;
9497
9598 this . scanInProgress = true ;
96- let lastScannedBlock = await this . getLastSeenBlockNumber ( ) ;
97- const currentBlockNumber = lastScannedBlock + 1 ;
98- let currentBlockHash = await this . blockchainService . getBlockHash ( currentBlockNumber ) ;
9999
100- if ( currentBlockHash . isEmpty ) {
101- this . logger . log ( 'No new blocks to read; no scan performed.' ) ;
102- this . scanInProgress = false ;
103- return ;
104- }
105- this . logger . log ( `Starting scan from block #${ currentBlockNumber } ${ currentBlockHash } ) ;
100+ let first = true ;
101+ while ( true ) {
102+ if ( this . paused ) {
103+ this . logger . log ( 'Scan paused' ) ;
104+ break ;
105+ }
106+
107+ const queueSize = await this . ipfsQueue . count ( ) ;
108+ if ( queueSize > this . configService . queueHighWater ) {
109+ this . logger . log ( 'Queue soft limit reached; pausing scan until next interval' ) ;
110+ break ;
111+ }
106112
107- while ( ! this . paused && ! currentBlockHash . isEmpty && queueSize < this . configService . queueHighWater ) {
108- const messages = await this . chainEventProcessor . getMessagesInBlock ( lastScannedBlock , eventsToWatch ) ;
113+ const currentBlockNumber = await this . getNextBlockNumber ( ) ;
114+ const currentBlockHash = await this . blockchainService . getBlockHash ( currentBlockNumber ) ;
115+ if ( currentBlockHash . isEmpty ) {
116+ this . logger . log ( `No new blocks to scan @ block number ${ currentBlockNumber } ) ;
117+ break ;
118+ }
119+
120+ if ( first ) {
121+ this . logger . log ( `Starting scan @ block # ${ currentBlockNumber } ${ currentBlockHash } ) ;
122+ first = false ;
123+ }
124+
125+ const messages = await this . chainEventProcessor . getMessagesInBlock ( currentBlockNumber , eventsToWatch ) ;
109126 if ( messages . length > 0 ) {
110127 this . logger . debug ( `Found ${ messages . length } ) ;
111128 }
112129 await this . chainEventProcessor . queueIPFSJobs ( messages , this . ipfsQueue ) ;
113- await this . saveProgress ( lastScannedBlock ) ;
114- lastScannedBlock += 1 ;
115- currentBlockHash = await this . blockchainService . getBlockHash ( lastScannedBlock ) ;
116- queueSize = await this . ipfsQueue . count ( ) ;
117- }
118- if ( currentBlockHash . isEmpty ) {
119- this . logger . log ( `Scan reached end-of-chain at block ${ lastScannedBlock - 1 } ) ;
120- } else if ( queueSize > this . configService . queueHighWater ) {
121- this . logger . log ( 'Queue soft limit reached; pausing scan until next iteration' ) ;
130+ await this . saveProgress ( currentBlockNumber ) ;
122131 }
123132 } catch ( err ) {
124133 this . logger . error ( err ) ;
@@ -127,8 +136,17 @@ export class ScannerService implements OnApplicationBootstrap {
127136 }
128137 }
129138
130- private async getLastSeenBlockNumber ( ) : Promise < number > {
131- return Number ( ( await this . cache . get ( LAST_SEEN_BLOCK_NUMBER_SCANNER_KEY ) ) ?? 0 ) ;
139+ private async getNextBlockNumber ( ) : Promise < number > {
140+ let nextBlock : number ;
141+ if ( this . scanResetBlockNumber ) {
142+ await this . setLastSeenBlockNumber ( this . scanResetBlockNumber - 1 ) ;
143+ nextBlock = this . scanResetBlockNumber ;
144+ this . scanResetBlockNumber = undefined ;
145+ } else {
146+ nextBlock = ( Number ( await this . cache . get ( LAST_SEEN_BLOCK_NUMBER_SCANNER_KEY ) ) ?? 0 ) + 1 ;
147+ }
148+
149+ return nextBlock ;
132150 }
133151
134152 private async saveProgress ( blockNumber : number ) : Promise < void > {
0 commit comments