import { Blockchain , Block , BlockBody , BlockHeader , Transaction , Consensus } from '@uniqys/blockchain' ; import { Network , NetworkOptions , SyncProtocol , SyncProtocolMeta , Message } from '@uniqys/p2p-network' ; import * as dapi from '@uniqys/dapp-interface' ; import { AsyncLoop } from '@uniqys/async-loop' ; import { EventEmitter } from 'events' ; import { Synchronizer } from './synchronizer' ; import { TransactionPool } from './transaction-pool' ; import { RemoteNodeSet , RemoteNode } from './remote-node' ; import PeerInfo from 'peer-info' ; import PeerId from 'peer-id' ; import debug from 'debug' ; const logger = debug ( 'chain-core:node' ) ; export interface NodeOptions extends NetworkOptions { propagateRateExponent : number ; handshakeTimeout : number ; } export namespace NodeOptions { export const defaults = Object . assign ( { propagateRateExponent : 0.5 , handshakeTimeout : 100 } , NetworkOptions . defaults ) ; } export class Node < T extends dapi . Dapp > implements dapi . Core { public readonly blockchain : Blockchain ; public readonly transactionPool : TransactionPool ; protected readonly event = new EventEmitter ( ) ; protected lastAppState ? : dapi . AppState ; private readonly dapp : T ; private readonly network : Network ; private readonly remoteNode = new RemoteNodeSet ( ) ; private readonly options : NodeOptions ; private readonly executionLoop = new AsyncLoop ( async ( ) => { await this . executeBlockTransactions ( ) ; } ) ; private readonly synchronizer : Synchronizer ; constructor ( dapp , blockchain , peerInfo , options ? ) { this . dapp = dapp ; this . blockchain = blockchain ; this . options = Object . assign ( { } , NodeOptions . defaults , options ) ; this . executionLoop . on ( 'error' , err => this . event . emit ( 'error' , err ) ) ; this . network = new Network ( peerInfo , this . options ) ; this . network . onError ( err => { if ( err . message === 'underlying socket has been closed' ) { return ; } this . event . emit ( 'error' , err ) ; } ) ; this . network . addProtocol ( new SyncProtocolMeta ( { handshake : ( protocol , incoming ) => this . handshake ( protocol , incoming ) , hello : ( msg , protocol ) => this . hello ( msg , protocol ) , newTransaction : ( msg , protocol ) => this . newTransaction ( msg , protocol ) , newBlock : ( msg , protocol ) => this . newBlock ( msg , protocol ) , newBlockHeight : ( msg , protocol ) => this . newBlockHeight ( msg , protocol ) , getConsentedHeader : ( msg , protocol ) => this . getConsentedHeader ( msg , protocol ) , getHeaders : ( msg , protocol ) => this . getHeaders ( msg , protocol ) , getBodies : ( msg , protocol ) => this . getBodies ( msg , protocol ) } ) ) ; this . synchronizer = new Synchronizer ( this . blockchain , this . remoteNode , ( node ) => this . dropRemoteNode ( node ) , ( block , consensus ) => this . propagateBlock ( block , consensus ) ) ; this . synchronizer . onError ( err => this . event . emit ( 'error' , err ) ) ; this . transactionPool = new TransactionPool ( ( tx ) => this . dapp . validateTransaction ( tx ) , ( txs ) => this . dapp . selectTransactions ( txs ) , ( tx ) => this . propagateTransaction ( tx ) ) ; } public onError ( listener ) { this . event . on ( 'error' , listener ) ; } public onExecuted ( listener ) { this . event . on ( 'executed' , listener ) ; } public async start ( ) { await this . initialize ( ) ; this . executionLoop . start ( ) ; this . synchronizer . start ( ) ; await this . network . start ( ) ; } public async stop ( ) { for ( const node of this . remoteNode . nodes ( ) ) { node . syncProtocol . end ( ) ; } await this . network . stop ( ) ; this . synchronizer . stop ( ) ; this . executionLoop . stop ( ) ; } public async sendTransaction ( transaction ) { await this . transactionPool . add ( transaction ) ; } protected propagateBlock ( block , consensus ) { this . _propagateBlock ( block , consensus ) . catch ( err => this . event . emit ( 'error' , err ) ) ; } private getRemoteNode ( protocol ) { const node = this . remoteNode . get ( protocol . peerId ) ; if ( ! node ) throw new Error ( 'Remote node disappeared' ) ; return node ; } private async handshake ( protocol , incoming ) { logger ( 'handshake %s %s %s' , this . network . localPeer . id . toB58String ( ) , incoming ? '<--' : '-->' , protocol . peerId ) ; protocol . onError ( err => { if ( err . message === 'underlying socket has been closed' ) { return ; } this . event . emit ( 'error' , err ) ; } ) ; protocol . start ( ) ; await protocol . sendHello ( new Message . Hello ( ( await this . blockchain . height ) , this . blockchain . genesisBlock . hash ) ) ; setTimeout ( ( ) => { if ( ! this . remoteNode . get ( protocol . peerId ) ) { this . dropPeer ( protocol . peerId ) ; } } , this . options . handshakeTimeout ) ; } private hello ( msg , protocol ) { if ( msg . genesis . equals ( this . blockchain . genesisBlock . hash ) ) { logger ( 'success handshake with %s height: %d genesis: %s ' , protocol . peerId , msg . height , msg . genesis . buffer . toString ( 'hex' ) ) ; const node = new RemoteNode ( protocol . peerId , protocol , msg . height ) ; this . addRemoteNode ( node ) ; protocol . onEnd ( ( ) => { logger ( 'goodby %s' , node . peerId ) ; this . remoteNode . delete ( node ) ; } ) ; } else { this . dropPeer ( protocol . peerId ) ; } } private addRemoteNode ( node ) { this . remoteNode . add ( node ) ; this . synchronizer . newNode ( ) ; } private dropRemoteNode ( node ) { this . remoteNode . delete ( node ) ; this . dropPeer ( node . peerId ) ; } private dropPeer ( id ) { this . network . dropPeer ( PeerId . createFromB58String ( id ) ) ; } private newTransaction ( msg , _ ) { this . transactionPool . add ( msg . transaction ) . catch ( err => this . event . emit ( 'error' , err ) ) ; } private newBlock ( msg , protocol ) { const node = this . getRemoteNode ( protocol ) ; this . synchronizer . newBlock ( msg , node ) ; } private newBlockHeight ( msg , protocol ) { const node = this . getRemoteNode ( protocol ) ; this . synchronizer . newBlockHeight ( msg , node ) ; } private async getConsentedHeader ( msg , _ ) { const header = await this . blockchain . headerOf ( msg . height ) ; const consensus = await this . blockchain . consensusOf ( msg . height ) ; return new Message . ConsentedHeader ( header , consensus ) ; } private async getHeaders ( msg , _ ) { const i = msg . from ; const last = Math . min ( await this . blockchain . height , i + msg . count - 1 ) ; const headers = [ ] ; for ( let i = msg . from ; i <= last ; i ++ ) { headers . push ( await this . blockchain . headerOf ( i ) ) ; } return new Message . Headers ( headers ) ; } private async getBodies ( msg , _ ) { const i = msg . from ; const last = Math . min ( await this . blockchain . height , i + msg . count - 1 ) ; const bodies = [ ] ; for ( let i = msg . from ; i <= last ; i ++ ) { bodies . push ( await this . blockchain . bodyOf ( i ) ) ; } return new Message . Bodies ( bodies ) ; } private async initialize ( ) { const [ height , appState ] = await Promise . all ( [ this . blockchain . ready ( ) . then ( ( ) => { return this . blockchain . height ; } ) , this . dapp . connect ( ) ] ) ; this . lastAppState = appState ; if ( this . lastAppState . height > height ) throw new Error ( 'need reset app' ) ; if ( this . lastAppState . height < height ) { const expect = ( await this . blockchain . headerOf ( this . lastAppState . height + 1 ) ) . appStateHash ; if ( ! this . lastAppState . hash . equals ( expect ) ) throw new Error ( 'app hash mismatch' ) ; } logger ( `initialized at block(${ this . lastAppState . height })` ) ; } private async executeBlockTransactions ( ) { if ( this . lastAppState === undefined ) { throw new Error ( 'not initialized' ) ; } if ( ( await this . blockchain . height ) !== this . lastAppState . height ) { const height = this . lastAppState . height + 1 ; const knownHeight = await this . blockchain . height ; logger ( `execute transaction in block(${ height })` ) ; const block = await this . blockchain . blockOf ( height ) ; const txs = block . body . transactionList . transactions ; const appState = await this . dapp . executeTransactions ( txs ) ; if ( appState . height !== height ) throw new Error ( 'block height mismatch' ) ; if ( appState . height < knownHeight ) { const expect = ( await this . blockchain . headerOf ( appState . height + 1 ) ) . appStateHash ; if ( ! this . lastAppState . hash . equals ( expect ) ) throw new Error ( 'app hash mismatch' ) ; } this . lastAppState = appState ; await this . transactionPool . update ( txs ) ; this . event . emit ( 'executed' , height ) ; } } 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 . syncProtocol . sendNewBlock ( newBlockMsg ) ) ) ; await Promise . all ( this . remoteNode . pickBlockReceivers ( height ) . map ( node => node . syncProtocol . sendNewBlockHeight ( newBlockHeightMsg ) ) ) ; } private propagateTransaction ( tx ) { this . _propagateTransaction ( tx ) . catch ( err => this . event . emit ( 'error' , err ) ) ; } private async _propagateTransaction ( tx ) { logger ( 'propagate transaction %s' , tx . hash . toHexString ( ) ) ; const newTransactionMsg = new Message . NewTransaction ( tx ) ; await Promise . all ( this . remoteNode . pickTransactionReceivers ( this . options . propagateRateExponent ) . map ( node => node . syncProtocol . sendNewTransaction ( newTransactionMsg ) ) ) ; } }