import * as AsyncLock from 'async-lock' ; import { getLogger , hexStrToBuf , hexBufToStr , logError , SignatureVerifier , DefaultSignatureVerifier , serializeObject , serializeObjectAsHexString , SignatureProvider , } from '@pigi/core-utils' ; import { DB , EthereumListener , EthereumEvent } from '@pigi/core-db' ; import { UnipigAggregator } from '../types/unipig-aggregator' ; import { Address , isFaucetTransaction , isSwapTransaction , NotSyncedError , RollupBlock , RollupBlockSubmitter , RollupStateMachine , RollupTransaction , RollupTransition , Signature , SignedStateReceipt , SignedTransaction , StateReceipt , StateSnapshot , StateUpdate , SwapTransition , Transfer , TransferTransition , } from '../types' ; import { abiEncodeStateReceipt , abiEncodeTransaction , abiEncodeTransition , parseTransitionFromABI , } from '../common/serialization' ; import { EMPTY_AGGREGATOR_SIGNATURE , generateTransferTx , PIGI_TOKEN_TYPE , UNI_TOKEN_TYPE , UNISWAP_ADDRESS , } from '../common' ; const log = getLogger ( 'rollup-aggregator' ) ; export class RollupAggregator implements EthereumListener < EthereumEvent > , UnipigAggregator { public static readonly PENDING_BLOCK_KEY : Buffer = Buffer . from ( 'pending_block_number' ) ; public static readonly LAST_TRANSITION_KEY : Buffer = Buffer . from ( 'last_transition' ) ; public static readonly TRANSACTION_COUNT_KEY : Buffer = Buffer . from ( 'tx_count' ) ; public static readonly TX_COUNT_STORAGE_THRESHOLD : number = 7 ; private static readonly lockKey : string = 'lock' ; private readonly lock : AsyncLock ; private synced : boolean ; private transactionCount : number ; private pendingBlock : RollupBlock ; private lastBlockSubmission : Date ; public static async create ( db , rollupStateMachine , rollupBlockSubmitter , signatureProvider , signatureVerifier = DefaultSignatureVerifier . instance ( ) , blockSubmissionTransitionCount = 100 , blockSubmissionIntervalMillis = 300000 , authorizedFaucetAddress ? ) { const aggregator = new RollupAggregator ( db , rollupStateMachine , rollupBlockSubmitter , signatureProvider , signatureVerifier , blockSubmissionTransitionCount , blockSubmissionIntervalMillis , authorizedFaucetAddress ) ; await aggregator . init ( ) ; return aggregator ; } private constructor ( private readonly db , private readonly rollupStateMachine , private readonly rollupBlockSubmitter , private readonly signatureProvider , private readonly signatureVerifier = DefaultSignatureVerifier . instance ( ) , private readonly blockSubmissionTransitionCount = 100 , private readonly blockSubmissionIntervalMillis = 300000 , private readonly authorizedFaucetAddress ? ) { this . pendingBlock = { blockNumber : 1 , transitions : [ ] , } ; this . lock = new AsyncLock ( ) ; this . synced = false ; } /**
     * Initialize method, required for the Aggregator to load existing state before
     * it can handle requests.
     */ private async init ( ) { try { const [ pendingBlockNumberBuffer , lastTransitionBuffer , txCountBuffer , ] = await Promise . all ( [ this . db . get ( RollupAggregator . PENDING_BLOCK_KEY ) , this . db . get ( RollupAggregator . LAST_TRANSITION_KEY ) , this . db . get ( RollupAggregator . TRANSACTION_COUNT_KEY ) , ] ) ; if ( ! lastTransitionBuffer ) { log . info ( `Init returning -- no stored last transition.` ) ; this . transactionCount = 0 ; this . lastBlockSubmission = new Date ( ) ; return ; } this . transactionCount = txCountBuffer ? parseInt ( txCountBuffer . toString ( ) , 10 ) : 0 ; const pendingBlock = pendingBlockNumberBuffer ? parseInt ( pendingBlockNumberBuffer . toString ( ) , 10 ) : 1 ; const lastTransition = parseInt ( lastTransitionBuffer . toString ( ) , 10 ) ; const promises = [ ] ; for ( let i = 1 ; i <= lastTransition ; i ++ ) { promises . push ( this . db . get ( RollupAggregator . getTransitionKey ( i ) ) ) ; } const transitionBuffers = await Promise . all ( promises ) ; const transitions = transitionBuffers . map ( ( x ) => parseTransitionFromABI ( hexBufToStr ( x ) ) ) ; this . pendingBlock = { blockNumber : pendingBlock , transitions , } ; log . info ( `Initialized aggregator with pending block: ${ JSON . stringify ( this . pendingBlock ) }` ) ; this . lastBlockSubmission = new Date ( ) ; this . setBlockSubmissionTimeout ( ) ; } catch ( e ) { logError ( log , 'Error initializing aggregator' , e ) ; throw e ; } } public async onSyncCompleted ( syncIdentifier ? ) { this . synced = true ; } public async handle ( event ) { log . debug ( `Aggregator received event: ${ JSON . stringify ( event ) }` ) ; if ( ! ! event && ! ! event . values && 'blockNumber' in event . values ) { await this . rollupBlockSubmitter . handleNewRollupBlock ( ( event . values [ 'blockNumber' ] as any ) . toNumber ( ) ) ; } } public async getTransactionCount ( ) { return this . transactionCount ; } public async getState ( address ) { if ( ! this . synced ) { throw new NotSyncedError ( ) ; } try { const stateReceipt = await this . lock . acquire ( RollupAggregator . lockKey , async ( ) => { const snapshot = await this . rollupStateMachine . getState ( address ) ; return { blockNumber : this . pendingBlock . blockNumber , transitionIndex : this . pendingBlock . transitions . length , ... snapshot , } ; } ) ; let signature ; if ( ! ! stateReceipt . state ) { signature = await this . signatureProvider . sign ( abiEncodeStateReceipt ( stateReceipt ) ) ; } else { signature = EMPTY_AGGREGATOR_SIGNATURE ; } return { stateReceipt , signature , } ; } catch ( e ) { log . error ( `Error getting state for address [${ address }]! ${ e . message }, ${ e . stack }` ) ; throw e ; } } public async applyTransaction ( signedTransaction ) { if ( ! this . synced ) { throw new NotSyncedError ( ) ; } try { const [ stateUpdate , blockNumber , transitionIndex , ] = await this . lock . acquire ( RollupAggregator . lockKey , async ( ) => { const update = await this . rollupStateMachine . applyTransaction ( signedTransaction ) ; await this . addToPendingBlock ( [ update ] , signedTransaction ) ; return [ update , this . pendingBlock . blockNumber , this . pendingBlock . transitions . length , ] ; } ) ; if ( this . pendingBlock . transitions . length >= this . blockSubmissionTransitionCount ) { this . submitBlock ( ) ; } await this . incrementTxCount ( ) ; return this . respond ( stateUpdate , blockNumber , transitionIndex ) ; } catch ( e ) { log . error ( `Error applying transaction [${ serializeObject ( signedTransaction ) }]! ${ e . message }, ${ e . stack }` ) ; throw e ; } } public async requestFaucetFunds ( signedTransaction ) { if ( ! this . synced ) { throw new NotSyncedError ( ) ; } try { if ( ! isFaucetTransaction ( signedTransaction . transaction ) ) { throw Error ( 'Cannot handle non-Faucet Request in faucet endpoint' ) ; } const messageSigner = this . signatureVerifier . verifyMessage ( serializeObjectAsHexString ( signedTransaction . transaction ) , signedTransaction . signature ) ; const requiredSigner = ! ! this . authorizedFaucetAddress ? this . authorizedFaucetAddress : signedTransaction . transaction . sender ; if ( messageSigner !== requiredSigner ) { throw Error ( `Faucet requests must be signed by the authorized faucet requester address. Signer address: ${ messageSigner }, required address: ${ requiredSigner }` ) ; } const { sender , amount } = signedTransaction . transaction ; const faucetTxs = await this . generateFaucetTxs ( sender , amount ) ; const [ stateUpdate , blockNumber , transitionIndex , ] = await this . lock . acquire ( RollupAggregator . lockKey , async ( ) => { const updates = await this . rollupStateMachine . applyTransactions ( faucetTxs ) ; await this . addToPendingBlock ( updates , signedTransaction ) ; return [ updates [ updates . length - 1 ] , this . pendingBlock . blockNumber , this . pendingBlock . transitions . length , ] ; } ) ; if ( this . pendingBlock . transitions . length >= this . blockSubmissionTransitionCount ) { this . submitBlock ( ) ; } await this . incrementTxCount ( ) ; return ( await this . respond ( stateUpdate , blockNumber , transitionIndex ) ) [ 1 ] ; } catch ( e ) { log . error ( `Error handling faucet request [${ serializeObject ( signedTransaction ) }]! ${ e . message }, ${ e . stack }` ) ; throw e ; } } param stateUpdate param blockNumber param transitionIndex returns private async respond ( stateUpdate , blockNumber , transitionIndex ) { const receipts = [ ] ; const senderReceipt = { slotIndex : stateUpdate . senderSlotIndex , stateRoot : stateUpdate . stateRoot , state : stateUpdate . senderState , inclusionProof : stateUpdate . senderStateInclusionProof , blockNumber , transitionIndex , } ; const senderSignature = await this . signatureProvider . sign ( abiEncodeStateReceipt ( senderReceipt ) ) ; receipts . push ( { signature : senderSignature , stateReceipt : senderReceipt , } ) ; if ( stateUpdate . receiverState . pubkey !== UNISWAP_ADDRESS ) { const recipientReceipt = { slotIndex : stateUpdate . receiverSlotIndex , stateRoot : stateUpdate . stateRoot , state : stateUpdate . receiverState , inclusionProof : stateUpdate . receiverStateInclusionProof , blockNumber , transitionIndex , } ; const recipientSignature = await this . signatureProvider . sign ( abiEncodeStateReceipt ( recipientReceipt ) ) ; receipts . push ( { signature : recipientSignature , stateReceipt : recipientReceipt , } ) ; } log . debug ( `Returning receipts: ${ serializeObject ( receipts ) }` ) ; return receipts ; } param updates param transaction returns private async addToPendingBlock ( updates , transaction ) { const transitions = [ ] ; if ( isSwapTransaction ( transaction . transaction ) ) { const update = updates [ 0 ] ; const transition = { stateRoot : update . stateRoot , senderSlotIndex : update . senderSlotIndex , uniswapSlotIndex : update . receiverSlotIndex , tokenType : transaction . transaction . tokenType , inputAmount : transaction . transaction . inputAmount , minOutputAmount : transaction . transaction . minOutputAmount , timeout : transaction . transaction . timeout , signature : transaction . signature , } ; this . pendingBlock . transitions . push ( transition ) ; await this . db . put ( RollupAggregator . getTransitionKey ( this . pendingBlock . transitions . length ) , hexStrToBuf ( abiEncodeTransition ( transition ) ) ) ; } else { for ( const u of updates ) { const transition = this . getTransferTransitionFromStateUpdate ( u ) ; this . pendingBlock . transitions . push ( transition ) ; await this . db . put ( RollupAggregator . getTransitionKey ( this . pendingBlock . transitions . length ) , hexStrToBuf ( abiEncodeTransition ( transition ) ) ) ; } } await this . db . put ( RollupAggregator . LAST_TRANSITION_KEY , Buffer . from ( this . pendingBlock . transitions . length . toString ( 10 ) ) ) ; return transitions ; } param update returns private getTransferTransitionFromStateUpdate ( update ) { const transfer = update . transaction . transaction as Transfer ; const transition = { stateRoot : update . stateRoot , senderSlotIndex : update . senderSlotIndex , recipientSlotIndex : update . receiverSlotIndex , tokenType : transfer . tokenType , amount : transfer . amount , signature : update . transaction . signature , } ; if ( update . receiverCreated ) { transition [ 'createdAccountPubkey' ] = update . receiverState . pubkey ; } return transition ; } /**
     * Submits a block to the main chain, creating a new pending block for future
     * transitions.
     */ private async submitBlock ( ) { return this . lock . acquire ( RollupAggregator . lockKey , async ( ) => { if ( this . pendingBlock . transitions . length < this . blockSubmissionTransitionCount ) { const millisSinceLastSubmission = new Date ( ) . getTime ( ) - this . lastBlockSubmission . getTime ( ) ; if ( millisSinceLastSubmission < this . blockSubmissionIntervalMillis ) { this . setBlockSubmissionTimeout ( this . blockSubmissionIntervalMillis - millisSinceLastSubmission ) ; return ; } else if ( this . pendingBlock . transitions . length === 0 ) { this . setBlockSubmissionTimeout ( this . blockSubmissionIntervalMillis ) ; return ; } } const toSubmit = this . pendingBlock ; await this . rollupBlockSubmitter . submitBlock ( toSubmit ) ; this . pendingBlock = { blockNumber : toSubmit . blockNumber + 1 , transitions : [ ] , } ; await this . db . put ( RollupAggregator . LAST_TRANSITION_KEY , Buffer . from ( '0' ) ) ; await this . db . put ( RollupAggregator . PENDING_BLOCK_KEY , Buffer . from ( this . pendingBlock . blockNumber . toString ( 10 ) ) ) ; this . lastBlockSubmission = new Date ( ) ; this . setBlockSubmissionTimeout ( ) ; } ) ; } private setBlockSubmissionTimeout ( timeoutMillis ? ) { setTimeout ( async ( ) => { await this . submitBlock ( ) ; } , timeoutMillis || this . blockSubmissionIntervalMillis ) ; } public static getTransitionKey ( transIndex ) { return Buffer . from ( `TRANS_${ transIndex }` ) ; } param recipient param amount returns private async generateFaucetTxs ( recipient , amount ) { const address = await this . signatureProvider . getAddress ( ) ; const txOne = generateTransferTx ( address , recipient , UNI_TOKEN_TYPE , amount ) ; const txTwo = generateTransferTx ( address , recipient , PIGI_TOKEN_TYPE , amount ) ; return [ { signature : await this . signatureProvider . sign ( abiEncodeTransaction ( txOne ) ) , transaction : txOne , } , { signature : await this . signatureProvider . sign ( abiEncodeTransaction ( txTwo ) ) , transaction : txTwo , } , ] ; } /**
     * Increments the total transaction count that this aggregator has processed,
     * saving periodically
     */ private async incrementTxCount ( ) { if ( ++ this . transactionCount % RollupAggregator . TX_COUNT_STORAGE_THRESHOLD === 0 ) { try { await this . db . put ( RollupAggregator . TRANSACTION_COUNT_KEY , Buffer . from ( this . transactionCount . toString ( 10 ) ) ) ; } catch ( e ) { logError ( log , 'Error saving transaction count!' , e ) ; } } } /***********
     * GETTERS *
     ***********/ public getPendingBlockNumber ( ) { return this . pendingBlock . blockNumber ; } public getNextTransitionIndex ( ) { return this . pendingBlock . transitions . length ; } }