import { BigNumber , getLogger , hexBufToStr , logError } from '@pigi/core-utils' ; import { DB , SparseMerkleTree , SparseMerkleTreeImpl } from '@pigi/core-db' ; import { RollupBlock , TransactionResult , TransactionStorage , } from '@pigi/rollup-core' ; import AsyncLock from 'async-lock' ; import { RollupBlockBuilder , RollupBlockSubmitter , TreeUpdateError , } from '../types' ; const parseTransactionResultFromABI = ( txResult ) : TransactionResult => { const json = JSON . parse ( txResult ) ; return { transactionNumber : new BigNumber ( json [ 'transactionNumber' ] , 'hex' ) , signedTransaction : json [ 'signedTransaction' ] , modifiedStorage : json [ 'modifiedStorage' ] , } ; } ; const abiEncodeTransactionResult = ( txResult ) : string => { return JSON . stringify ( { transactionNumber : txResult . transactionNumber . toString ( 'hex' ) , signedTransaction : txResult . signedTransaction , modifiedStorage : txResult . modifiedStorage , } ) ; } ; const log = getLogger ( 'rollup-block-builder' ) ; interface PendingBlock { blockNumber : number ; transactionResults : TransactionResult [ ] ; } /**
 *  Default Block Builder implementation. We don't expect others.
 */ export class DefaultRollupBlockBuilder implements RollupBlockBuilder { public static readonly LAST_SUBMISSION_DATE_KEY : Buffer = Buffer . from ( 'last_submission' ) ; public static readonly PENDING_BLOCK_KEY : Buffer = Buffer . from ( 'pending_block_number' ) ; public static readonly TRANSACTION_COUNT_KEY : Buffer = Buffer . from ( 'tx_count' ) ; public static readonly TREE_ROOT_KEY : Buffer = Buffer . from ( 'tree_root' ) ; private static readonly LOCK_KEY : string = 'lock' ; private readonly lock : AsyncLock ; private tree : SparseMerkleTree ; private subtrees : SparseMerkleTree [ ] ; private lastBlockSubmission : Date ; private pendingBlock : PendingBlock ; public static async create ( db , rollupBlockSubmitter , maxTransactionsPerBlock = 100 , maxDelayBetweenBlocksMillis = 30000 ) { const blockBuilder = new DefaultRollupBlockBuilder ( db , rollupBlockSubmitter , maxTransactionsPerBlock , maxDelayBetweenBlocksMillis ) ; await blockBuilder . init ( ) ; return blockBuilder ; } constructor ( private readonly db , private readonly rollupBlockSubmitter , private readonly maxTransactionsPerBlock , private readonly maxDelayBetweenBlocksMillis ) { this . pendingBlock = { blockNumber : 0 , transactionResults : [ ] , } ; this . lock = new AsyncLock ( ) ; } private async init ( ) { try { const [ lastSubmissionDateBuffer , txCountBuffer , pendingBlockBuffer , treeRoot , ] = await Promise . all ( [ this . db . get ( DefaultRollupBlockBuilder . LAST_SUBMISSION_DATE_KEY ) , this . db . get ( DefaultRollupBlockBuilder . TRANSACTION_COUNT_KEY ) , this . db . get ( DefaultRollupBlockBuilder . PENDING_BLOCK_KEY ) , this . db . get ( DefaultRollupBlockBuilder . TREE_ROOT_KEY ) , ] ) ; if ( ! txCountBuffer ) { log . info ( `Init returning -- no stored last transition. This is a fresh start.` ) ; this . lastBlockSubmission = new Date ( ) ; this . setBlockSubmissionTimeout ( ) ; this . tree = await SparseMerkleTreeImpl . create ( this . db , undefined , 16 ) ; this . subtrees = [ ] ; return ; } this . lastBlockSubmission = ! ! lastSubmissionDateBuffer ? new Date ( parseInt ( lastSubmissionDateBuffer . toString ( ) , 10 ) ) : new Date ( ) ; const transactionCount = txCountBuffer ? parseInt ( txCountBuffer . toString ( ) , 10 ) : 0 ; const pendingBlock = pendingBlockBuffer ? parseInt ( pendingBlockBuffer . toString ( ) , 10 ) : 1 ; this . tree = await SparseMerkleTreeImpl . create ( this . db , treeRoot , 16 ) ; this . subtrees = [ ] ; const promises = [ ] ; for ( let i = 1 ; i <= transactionCount ; i ++ ) { promises . push ( this . db . get ( DefaultRollupBlockBuilder . getTransactionKey ( i ) ) ) ; } const transactionBuffers = await Promise . all ( promises ) ; const transactionResults = transactionBuffers . map ( ( x ) => parseTransactionResultFromABI ( x . toString ( ) ) ) ; this . pendingBlock = { blockNumber : pendingBlock , transactionResults , } ; log . info ( `Initialized aggregator with pending block: ${ JSON . stringify ( this . pendingBlock ) } and tree root: ${ ( await this . tree . getRootHash ( ) ) . toString ( 'hex' ) }` ) ; return this . submitBlock ( ) ; } catch ( e ) { logError ( log , 'Error initializing aggregator' , e ) ; throw e ; } } public async addTransactionResult ( transactionResult ) { if ( this . pendingBlock . transactionResults . length > 0 && this . pendingBlock . transactionResults [ this . pendingBlock . transactionResults . length - 1 ] . transactionNumber . eq ( transactionResult . transactionNumber ) ) { log . warn ( `Ignoring duplicate TransactionResult. Received [${ JSON . stringify ( transactionResult ) }], 
        but last transaction is 
        [${ this . pendingBlock . transactionResults [ this . pendingBlock . transactionResults . length - 1 ] }].` ) ; return ; } this . pendingBlock . transactionResults . push ( transactionResult ) ; log . debug ( `Received TransactionResult [${ JSON . stringify ( transactionResult ) }]. Pending block [${ this . pendingBlock . blockNumber }] size: [${ this . pendingBlock . transactionResults . length }]` ) ; await this . db . put ( DefaultRollupBlockBuilder . getTransactionKey ( this . pendingBlock . transactionResults . length ) , Buffer . from ( abiEncodeTransactionResult ( transactionResult ) ) ) ; await this . db . put ( DefaultRollupBlockBuilder . TRANSACTION_COUNT_KEY , Buffer . from ( this . pendingBlock . transactionResults . length . toString ( 10 ) ) ) ; if ( this . pendingBlock . transactionResults . length >= this . maxTransactionsPerBlock ) { log . debug ( `Submitting block [${ this . pendingBlock . blockNumber }]` ) ; return this . submitBlock ( ) ; } else { log . debug ( `Not submitting partial block. Pending block [${ this . pendingBlock . blockNumber }] is at ${ this . pendingBlock . transactionResults . length }/${ this . maxTransactionsPerBlock } of its capacity.` ) ; } } /**
     * Submits a block to the main chain, creating a new pending block for future
     * transitions.
     */ private async submitBlock ( ) { log . debug ( `Waiting to acquire lock to submit block [${ this . pendingBlock . blockNumber }]` ) ; return this . lock . acquire ( DefaultRollupBlockBuilder . LOCK_KEY , async ( ) => { log . debug ( `Lock acquired to submit block [${ this . pendingBlock . blockNumber }]` ) ; if ( this . pendingBlock . transactionResults . length < this . maxTransactionsPerBlock ) { const millisSinceLastSubmission = new Date ( ) . getTime ( ) - this . lastBlockSubmission . getTime ( ) ; if ( millisSinceLastSubmission < this . maxDelayBetweenBlocksMillis ) { log . debug ( `Not submitting block: Block tx count [${ Object . keys ( this . pendingBlock . transactionResults ) . length }] less than max per block [${ this . maxTransactionsPerBlock }], and max time between blocks has not elapsed.` ) ; this . setBlockSubmissionTimeout ( this . maxDelayBetweenBlocksMillis - millisSinceLastSubmission ) ; return ; } else if ( this . pendingBlock . transactionResults . length === 0 ) { log . info ( `Not submitting block: Block is empty.` ) ; this . setBlockSubmissionTimeout ( this . maxDelayBetweenBlocksMillis ) ; return ; } } log . info ( `Building block # [${ this . pendingBlock . blockNumber }]` ) ; const toSubmit = await this . buildBlock ( ) ; log . info ( `Block built. Submitting block # [${ toSubmit . blockNumber }]` ) ; await this . rollupBlockSubmitter . submitBlock ( toSubmit ) ; log . info ( `Block # [${ toSubmit . blockNumber }] submitted.` ) ; await this . db . put ( DefaultRollupBlockBuilder . TRANSACTION_COUNT_KEY , Buffer . from ( '0' ) ) ; await this . db . put ( DefaultRollupBlockBuilder . PENDING_BLOCK_KEY , Buffer . from ( this . pendingBlock . blockNumber . toString ( 10 ) ) ) ; this . lastBlockSubmission = new Date ( ) ; this . setBlockSubmissionTimeout ( ) ; } ) ; } returns private async buildBlock ( ) { log . debug ( `Building Block to submit. Block # [${ this . pendingBlock . blockNumber }]` ) ; const block = this . pendingBlock ; this . pendingBlock = { blockNumber : block . blockNumber + 1 , transactionResults : [ ] , } ; const modifiedStorageMap = new Map < string , TransactionStorage > ( ) ; for ( const res of block . transactionResults ) { for ( const modifiedStorage of res . modifiedStorage ) { modifiedStorageMap . set ( `${ modifiedStorage . contractSlotIndex }_${ modifiedStorage . storageSlotIndex }` , modifiedStorage ) ; } } const modifiedContractSlotIndexes = new Set ( ) ; const storagePromises = [ ] ; for ( const modifiedStorage of modifiedStorageMap . values ( ) ) { storagePromises . push ( this . updateStorageSlot ( modifiedStorage ) ) ; modifiedContractSlotIndexes . add ( modifiedStorage . contractSlotIndex ) ; } log . debug ( `Awaiting updateStorageSlot promises. Count: [${ storagePromises . length }]` ) ; await Promise . all ( storagePromises ) ; log . debug ( `updateStorageSlot promises completed` ) ; const blockPromises = [ ] ; for ( const idx of modifiedContractSlotIndexes . keys ( ) ) { blockPromises . push ( this . updateContractSlot ( idx ) ) ; } log . debug ( `Awaiting updateContractSlot promises. Count: [${ blockPromises . length }]` ) ; await Promise . all ( blockPromises ) ; log . debug ( `updateContractSlot completed` ) ; const stateRoot = await this . tree . getRootHash ( ) ; return { blockNumber : block . blockNumber , stateRoot : stateRoot . toString ( 'hex' ) , signedTransactions : block . transactionResults . map ( ( x ) => x . signedTransaction ) , } ; } param transactionStorage private async updateStorageSlot ( transactionStorage ) { log . debug ( `Updating contract storage [${ transactionStorage . contractSlotIndex }, ${ transactionStorage . storageSlotIndex }] to [${ transactionStorage . storage }].` ) ; const contractSlotBN = new BigNumber ( transactionStorage . contractSlotIndex , 10 ) ; const subtreeRoot = await this . tree . getLeaf ( contractSlotBN ) ; if ( ! subtreeRoot ) { log . debug ( `Creating contract slot index [${ transactionStorage . contractSlotIndex }].` ) ; this . subtrees [ transactionStorage . contractSlotIndex ] = await SparseMerkleTreeImpl . create ( this . db , undefined , 32 ) ; } else if ( ! this . subtrees [ transactionStorage . contractSlotIndex ] ) { log . info ( `Subtree at index [${ transactionStorage . contractSlotIndex }] exists with root [${ subtreeRoot . toString ( 'hex' ) }] but is not in the subtree array. Creating it.` ) ; this . subtrees [ transactionStorage . contractSlotIndex ] = await SparseMerkleTreeImpl . create ( this . db , subtreeRoot , 32 ) ; } const storageSlotBN = new BigNumber ( transactionStorage . storageSlotIndex , 10 ) ; let updated ; try { updated = await this . subtrees [ transactionStorage . contractSlotIndex ] . update ( storageSlotBN , Buffer . from ( transactionStorage . storage ) ) ; } catch ( e ) { logError ( log , `Error updating contract storage [${ transactionStorage . contractSlotIndex }, ${ transactionStorage . storageSlotIndex }] to [${ transactionStorage . storage }].` , e ) ; throw e ; } if ( ! updated ) { const msg = `Error updating contract storage [${ transactionStorage . contractSlotIndex }, ${ transactionStorage . storageSlotIndex }] to [${ transactionStorage . storage }].` ; log . error ( msg ) ; throw new TreeUpdateError ( msg ) ; } log . debug ( `Updated contract storage [${ transactionStorage . contractSlotIndex }, ${ transactionStorage . storageSlotIndex }] to [${ transactionStorage . storage }].` ) ; } param contractSlotIndex private async updateContractSlot ( contractSlotIndex ) { log . debug ( `Updating contract slot index [${ contractSlotIndex }] with subtree hash.` ) ; const subtreeHash = await this . subtrees [ contractSlotIndex ] . getRootHash ( ) ; const slotBN = new BigNumber ( contractSlotIndex , 10 ) ; const updated = await this . tree . update ( slotBN , subtreeHash ) ; if ( ! updated ) { const msg = `Error updating contract slot index [${ contractSlotIndex }] with new tree root` ; log . error ( msg ) ; throw new TreeUpdateError ( msg ) ; } log . debug ( `Updated Contract Slot Index [${ contractSlotIndex }] to [${ subtreeHash . toString ( 'hex' ) }].` ) ; } param timeoutMillis private setBlockSubmissionTimeout ( timeoutMillis ? ) { setTimeout ( async ( ) => { await this . submitBlock ( ) ; } , timeoutMillis || this . maxDelayBetweenBlocksMillis ) ; } param txNumber returns private static getTransactionKey ( txNumber ) { return Buffer . from ( `tx${ txNumber . toString ( 10 ) }` ) ; } }