import { Blockchain , Block , BlockHeader , Consensus } from '@uniqys/blockchain' ; import { AsyncLoop } from '@uniqys/async-loop' ; import { Message } from '@uniqys/protocol' ; import { PriorityQueue , Utility } from '@uniqys/priority-queue' ; import { Mutex } from '@uniqys/lock' ; import { RemoteNode , RemoteNodeSet } from './remote-node' ; import { EventEmitter } from 'events' ; import debug from 'debug' ; const logger = debug ( 'chain-core:sync' ) ; export interface SynchronizerOptions { waitForFetch : number ; catchUpInterval : number ; maxFetchHeaders : number ; maxFetchBodies : number ; maxPendingGap : number ; propagateRateExponent : number ; } export namespace SynchronizerOptions { export const defaults = { waitForFetch : 500 , catchUpInterval : 1000 , maxFetchHeaders : 1000 , maxFetchBodies : 100 , maxPendingGap : 50 , propagateRateExponent : 0.5 } ; } export class Synchronizer { private readonly options : SynchronizerOptions ; private readonly blockQueue = new PriorityQueue ( Utility . ascending < { block : Block ; consensus : Consensus ; from ? : RemoteNode ; } > ( ) ) ; private readonly fetchSchedule = new Map < number , NodeJS . Timer > ( ) ; private readonly chainingLoop = new AsyncLoop ( ( ) => this . chainPendingBlock ( ) ) ; private readonly catchUpMutex = new Mutex ( ) ; private readonly event = new EventEmitter ( ) ; private catchUpTimer ? : NodeJS . Timer ; constructor ( private readonly blockchain , private readonly remoteNode , private readonly dropRemoteNode , options ? ) { this . options = Object . assign ( { } , SynchronizerOptions . defaults , options ) ; this . chainingLoop . on ( 'error' , err => this . event . emit ( 'error' , err ) ) ; } public start ( ) { this . chainingLoop . start ( ) ; this . resetCatchUpTimer ( ) ; } public stop ( ) { this . chainingLoop . stop ( ) ; this . clearCatchUpTimer ( ) ; } public onError ( listener ) { this . event . on ( 'error' , listener ) ; } public newBlockHeight ( message , from ) { if ( from . height < message . height ) { from . height = message . height ; this . scheduleFetch ( message . height ) . catch ( err => this . event . emit ( 'error' , err ) ) ; } } public newBlock ( message , from ) { if ( from . height < message . block . header . height ) { from . height = message . block . header . height ; } this . enqueueNewBlock ( message . block , message . consensus , from ) . catch ( err => this . event . emit ( 'error' , err ) ) ; } public newBlockFromLocal ( block , consensus ) { this . enqueueNewBlock ( block , consensus ) . catch ( err => this . event . emit ( 'error' , err ) ) ; } public newNode ( ) { this . catchUp ( ) ; } private async scheduleFetch ( height ) { if ( this . fetchSchedule . get ( height ) || height <= ( await this . blockchain . height ) ) { return ; } this . fetchSchedule . set ( height , setTimeout ( ( ) => { this . fetch ( height ) . catch ( err => { this . event . emit ( 'error' , err ) ; } ) ; } , this . options . waitForFetch ) ) ; } private async enqueueNewBlock ( block , consensus , from ? ) { const knownHeight = await this . blockchain . height ; const height = block . header . height ; if ( height <= knownHeight || ( height - knownHeight ) > this . options . maxPendingGap ) { return ; } this . enqueueBlock ( block , consensus , from ) ; } private enqueueBlock ( block , consensus , from ? ) { this . blockQueue . enqueue ( { priority : block . header . height , value : { block , consensus , from } } ) ; } private async fetch ( height ) { const knownHeight = await this . blockchain . height ; if ( height <= knownHeight || ( height - knownHeight ) > this . options . maxPendingGap ) { return ; } let _provider = this . remoteNode . pickIdleProvider ( height ) ; while ( _provider ) { const leader = _provider ; try { await leader . use ( async ( ) => { const consented = await leader . protocol . fetchConsentedHeader ( new Message . GetConsentedHeader ( height ) ) ; const body = ( await leader . protocol . fetchBodies ( new Message . GetBodies ( height , 1 ) ) ) . bodies [ 0 ] ; this . enqueueBlock ( new Block ( consented . header , body ) , consented . consensus , leader ) ; } ) ; return ; } catch ( err ) { logger ( 'fetch invalid from: %d' , leader . peerId ) ; this . dropRemoteNode ( leader ) ; } _provider = this . remoteNode . pickIdleProvider ( height ) ; } } private resetCatchUpTimer ( ) { this . clearCatchUpTimer ( ) ; this . catchUpTimer = setTimeout ( ( ) => { this . catchUp ( ) ; } , this . options . catchUpInterval ) ; } private clearCatchUpTimer ( ) { if ( this . catchUpTimer ) clearTimeout ( this . catchUpTimer ) ; } private catchUp ( ) { if ( ! this . catchUpMutex . locked ) { this . catchUpMutex . use ( ( ) => this . catchUpAsync ( ) ) . catch ( err => this . event . emit ( 'error' , err ) ) ; } this . resetCatchUpTimer ( ) ; } private async catchUpAsync ( ) { const best = this . remoteNode . bestNode ( ) ; if ( ! best ) { return ; } const known = await this . blockchain . height ; if ( best . height <= known ) { return ; } try { const consented = await this . fetchLastConsentedHeader ( best ) ; await this . catchUpHeaders ( best , consented . header ) ; await this . catchUpBodies ( consented . header . height ) ; await this . blockchain . blockStore . rwLock . writeLock . use ( ( ) => this . updateHeight ( consented . header . height , consented . consensus ) ) ; logger ( 'catch up with %s to height %d' , best . peerId , consented . header . height ) ; } catch ( err ) { logger ( 'catch up failed: %s' , err . message ) ; } } private async fetchLastConsentedHeader ( best ) { return best . use ( async ( ) => { const height = best . height ; const consented = await best . protocol . fetchConsentedHeader ( new Message . GetConsentedHeader ( height ) ) ; try { consented . consensus . validate ( consented . header . hash , this . blockchain . genesisBlock . hash , consented . validatorSet ) ; } catch ( err ) { this . dropRemoteNode ( best ) ; throw err ; } return consented ; } ) ; } private async catchUpHeaders ( provider , last ) { await this . blockchain . blockStore . setHeader ( last . height , last ) ; await provider . use ( async ( ) => { while ( true ) { const unknown = ( await this . blockchain . height ) + 1 ; if ( last . height <= unknown ) { break ; } const from = Math . max ( last . height - this . options . maxFetchHeaders , unknown ) ; const count = last . height - from ; const msg = await provider . protocol . fetchHeaders ( new Message . GetHeaders ( from , count ) ) ; const headers = msg . headers . reverse ( ) ; for ( const header of headers ) { if ( header . height + 1 !== last . height || ! header . hash . equals ( last . lastBlockHash ) ) { this . dropRemoteNode ( provider ) ; throw new Error ( 'fetched header is invalid' ) ; } await this . blockchain . blockStore . setHeader ( header . height , header ) ; last = header ; } } } ) ; logger ( 'catch up headers complete' ) ; } private async catchUpBodies ( target ) { const jobQueue = new PriorityQueue < { from : number ; to : number ; } > ( ( job1 , job2 ) => job2 . to - job1 . to ) ; const known = await this . blockchain . height ; for ( let to = target ; known < to ; to -= this . options . maxFetchBodies ) { const from = Math . max ( to - this . options . maxFetchBodies , known + 1 ) ; jobQueue . enqueue ( { from , to } ) ; } let inProgress = 0 ; await new Promise ( ( resolve , reject ) => { const fetchLoop = new AsyncLoop ( async ( ) => { const job = jobQueue . dequeue ( ) ; if ( ! job && inProgress === 0 ) { fetchLoop . stop ( ) ; return ; } if ( ! job ) { return ; } const provider = this . remoteNode . pickIdleProvider ( job . to ) ; if ( ! provider ) { return jobQueue . enqueue ( job ) ; } inProgress ++ ; provider . use ( async ( ) => { const from = job . from ; const count = job . to - job . from + 1 ; const msg = await provider . protocol . fetchBodies ( new Message . GetBodies ( from , count ) ) ; for ( let i = 0 ; i < count ; i ++ ) { const header = await this . blockchain . blockStore . getHeader ( from + i ) ; const body = msg . bodies [ i ] ; try { body . validate ( header ) ; } catch ( err ) { logger ( 'fetched body is invalid: %s' , err . message ) ; this . dropRemoteNode ( provider ) ; throw err ; } await this . blockchain . blockStore . setBody ( from + i , body ) ; } } ) . finally ( ( ) => { inProgress -- ; } ) . catch ( err => { logger ( 'retry fetch body from %d to %d. reason: %s' , job . from , job . to , err . message ) ; jobQueue . enqueue ( job ) ; } ) ; } ) ; fetchLoop . on ( 'error' , reject ) ; fetchLoop . on ( 'end' , resolve ) ; fetchLoop . start ( ) ; } ) ; logger ( 'catch up bodies complete' ) ; } private async updateHeight ( height , consensus ) { const known = await this . blockchain . height ; if ( height > known ) { await this . blockchain . blockStore . setHeight ( height ) ; await this . blockchain . blockStore . setLastConsensus ( consensus ) ; } } private async chainPendingBlock ( ) { const pendingOrUndef = this . blockQueue . dequeue ( ) ; if ( pendingOrUndef ) { const pending = pendingOrUndef . value ; try { const knownHeight = await this . blockchain . height ; const height = pending . block . header . height ; if ( height > knownHeight ) { if ( height === knownHeight + 1 ) { pending . block . validate ( ) ; pending . consensus . validate ( pending . block . hash , this . blockchain . genesisBlock . hash , await this . blockchain . validatorSetOf ( height ) ) ; await this . blockchain . blockStore . rwLock . writeLock . use ( async ( ) => { const knownHeight = await this . blockchain . height ; if ( height === knownHeight + 1 ) { await Promise . all ( [ this . blockchain . blockStore . setHeader ( height , pending . block . header ) , this . blockchain . blockStore . setBody ( height , pending . block . body ) ] ) ; await this . updateHeight ( height , pending . consensus ) ; } } ) ; if ( pending . from ) { logger ( 'chain block(%d) from remote: %s' , pending . block . header . height , pending . from . peerId ) ; } else { logger ( 'chain block(%d) from local' , pending . block . header . height ) ; } this . resetCatchUpTimer ( ) ; await this . propagateBlock ( pending . block , pending . consensus ) ; } else { this . enqueueBlock ( pending . block , pending . consensus , pending . from ) ; } } } catch ( err ) { if ( pending . from ) { logger ( 'failed to chain the pending block from remote: %s' , err . message ) ; this . dropRemoteNode ( pending . from ) ; } else { logger ( 'failed to chain the pending block from local: %s' , err . message ) ; throw new Error ( 'failed to chain from local' ) ; } } } } private async propagateBlock ( block , consensus ) { const height = block . header . height ; logger ( 'propagate block (%d)' , height ) ; const newBlockMsg = new Message . NewBlock ( block , consensus ) ; const newBlockHeightMsg = new Message . NewBlockHeight ( height ) ; await Promise . all ( this . remoteNode . pickBlockReceivers ( height , this . options . propagateRateExponent ) . map ( node => node . protocol . sendNewBlock ( newBlockMsg ) ) ) ; await Promise . all ( this . remoteNode . pickBlockReceivers ( height ) . map ( node => node . protocol . sendNewBlockHeight ( newBlockHeightMsg ) ) ) ; } }