import { getAddressFromPublicKey } from '@liskhq/lisk-cryptography' ; import * as createDebug from 'debug' ; import { EventEmitter } from 'events' ; import { dataStructures } from '@liskhq/lisk-utils' ; import { TransactionPoolError } from './errors' ; import { Job } from './job' ; import { TransactionList } from './transaction_list' ; import { Status , Transaction , TransactionStatus } from './types' ; const debug = createDebug ( 'lisk:transaction_pool' ) ; type ApplyFunction = ( transactions : ReadonlyArray < Transaction > ) => Promise < void > ; interface BaseFee { readonly moduleID : number ; readonly assetID : number ; readonly baseFee : bigint ; } export interface TransactionPoolConfig { readonly maxTransactions ? : number ; readonly maxTransactionsPerAccount ? : number ; readonly transactionExpiryTime ? : number ; readonly minEntranceFeePriority ? : bigint ; readonly transactionReorganizationInterval ? : number ; readonly minReplacementFeeDifference ? : bigint ; readonly minFeePerByte : number ; readonly baseFees : BaseFee [ ] ; applyTransactions ( transactions ) : Promise < void > ; } interface TransactionFailedResponse { readonly code : string ; readonly id : Buffer ; readonly transactionError : Error & { code : string ; } ; } interface AddTransactionResponse { readonly status : Status ; readonly error ? : Error ; } export const DEFAULT_MAX_TRANSACTIONS = 4096 ; export const DEFAULT_MAX_TRANSACTIONS_PER_ACCOUNT = 64 ; export const DEFAULT_MIN_ENTRANCE_FEE_PRIORITY = BigInt ( 0 ) ; export const DEFAULT_EXPIRY_TIME = 3 * 60 * 60 * 1000 ; export const DEFAULT_EXPIRE_INTERVAL = 60 * 60 * 1000 ; export const DEFAULT_MINIMUM_REPLACEMENT_FEE_DIFFERENCE = BigInt ( 10 ) ; export const DEFAULT_REORGANIZE_TIME = 500 ; export const events = { EVENT_TRANSACTION_REMOVED : 'EVENT_TRANSACTION_REMOVED' , } ; const ERR_NONCE_OUT_OF_BOUNDS_CODE = 'ERR_NONCE_OUT_OF_BOUNDS' ; const ERR_TRANSACTION_VERIFICATION_FAIL = 'ERR_TRANSACTION_VERIFICATION_FAIL' ; export class TransactionPool { public events : EventEmitter ; private readonly _allTransactions : dataStructures . BufferMap < Transaction > ; private readonly _transactionList : dataStructures . BufferMap < TransactionList > ; private readonly _applyFunction : ApplyFunction ; private readonly _maxTransactions : number ; private readonly _maxTransactionsPerAccount : number ; private readonly _transactionExpiryTime : number ; private readonly _minEntranceFeePriority : bigint ; private readonly _transactionReorganizationInterval : number ; private readonly _minReplacementFeeDifference : bigint ; private readonly _minFeePerByte : number ; private readonly _baseFees : BaseFee [ ] ; private readonly _reorganizeJob : Job < void > ; private readonly _feePriorityQueue : dataStructures . MinHeap < Buffer , bigint > ; private readonly _expireJob : Job < void > ; public constructor ( config ) { this . events = new EventEmitter ( ) ; this . _feePriorityQueue = new dataStructures . MinHeap < Buffer , bigint > ( ) ; this . _allTransactions = new dataStructures . BufferMap < Transaction > ( ) ; this . _transactionList = new dataStructures . BufferMap < TransactionList > ( ) ; this . _applyFunction = config . applyTransactions ; this . _maxTransactions = config . maxTransactions ?? DEFAULT_MAX_TRANSACTIONS ; this . _maxTransactionsPerAccount = config . maxTransactionsPerAccount ?? DEFAULT_MAX_TRANSACTIONS_PER_ACCOUNT ; this . _transactionExpiryTime = config . transactionExpiryTime ?? DEFAULT_EXPIRY_TIME ; this . _minEntranceFeePriority = config . minEntranceFeePriority ?? DEFAULT_MIN_ENTRANCE_FEE_PRIORITY ; this . _transactionReorganizationInterval = config . transactionReorganizationInterval ?? DEFAULT_REORGANIZE_TIME ; this . _minReplacementFeeDifference = config . minReplacementFeeDifference ?? DEFAULT_MINIMUM_REPLACEMENT_FEE_DIFFERENCE ; this . _baseFees = config . baseFees ; this . _minFeePerByte = config . minFeePerByte ; this . _reorganizeJob = new Job ( async ( ) => this . _reorganize ( ) , this . _transactionReorganizationInterval ) ; this . _expireJob = new Job ( async ( ) => this . _expire ( ) , DEFAULT_EXPIRE_INTERVAL ) ; } public async start ( ) { this . _reorganizeJob . start ( ) ; this . _expireJob . start ( ) ; } public stop ( ) { this . _reorganizeJob . stop ( ) ; this . _expireJob . stop ( ) ; } public getAll ( ) { return this . _allTransactions . values ( ) ; } public get ( id ) { return this . _allTransactions . get ( id ) ; } public contains ( id ) { return this . _allTransactions . has ( id ) ; } public async add ( incomingTx ) { if ( this . _allTransactions . has ( incomingTx . id ) ) { debug ( 'Received duplicate transaction' , incomingTx . id ) ; return { status : Status . OK } ; } incomingTx . feePriority = this . _calculateFeePriority ( incomingTx ) ; if ( incomingTx . feePriority < this . _minEntranceFeePriority ) { const error = new TransactionPoolError ( 'Rejecting transaction due to failed minimum entrance fee priority requirement.' , incomingTx . id , '.fee' , incomingTx . feePriority . toString ( ) , this . _minEntranceFeePriority . toString ( ) ) ; return { status : Status . FAIL , error } ; } const lowestFeePriorityTrx = this . _feePriorityQueue . peek ( ) ; if ( this . _allTransactions . size >= this . _maxTransactions && lowestFeePriorityTrx && incomingTx . feePriority <= lowestFeePriorityTrx . key ) { const error = new TransactionPoolError ( 'Rejecting transaction due to fee priority when the pool is full.' , incomingTx . id , '.fee' , incomingTx . feePriority . toString ( ) , lowestFeePriorityTrx . key . toString ( ) ) ; return { status : Status . FAIL , error } ; } const incomingTxAddress = getAddressFromPublicKey ( incomingTx . senderPublicKey ) ; let txStatus ; try { await this . _applyFunction ( [ incomingTx ] ) ; txStatus = TransactionStatus . PROCESSABLE ; } catch ( err ) { txStatus = this . _getStatus ( err ) ; if ( txStatus === TransactionStatus . INVALID ) { return { status : Status . FAIL , error : err as Error , } ; } } const exceededTransactionsCount = this . _allTransactions . size - this . _maxTransactions ; if ( exceededTransactionsCount >= 0 ) { const isEvicted = this . _evictUnprocessable ( ) ; if ( ! isEvicted ) { this . _evictProcessable ( ) ; } } if ( ! this . _transactionList . has ( incomingTxAddress ) ) { this . _transactionList . set ( incomingTxAddress , new TransactionList ( incomingTxAddress , { maxSize : this . _maxTransactionsPerAccount , minReplacementFeeDifference : this . _minReplacementFeeDifference , } ) ) ; } const { added , removedID , reason } = ( this . _transactionList . get ( incomingTxAddress ) as TransactionList ) . add ( incomingTx , txStatus === TransactionStatus . PROCESSABLE ) ; if ( ! added ) { return { status : Status . FAIL , error : new TransactionPoolError ( reason , incomingTx . id ) , } ; } if ( removedID ) { debug ( 'Removing from transaction pool with id' , removedID ) ; const removedTx = this . _allTransactions . get ( removedID ) as Transaction ; this . _allTransactions . delete ( removedID ) ; this . events . emit ( events . EVENT_TRANSACTION_REMOVED , { id : removedTx . id , nonce : removedTx . nonce . toString ( ) , senderPublicKey : removedTx . senderPublicKey , reason : 'Transaction List executed remove' , } ) ; } incomingTx . receivedAt = new Date ( ) ; this . _allTransactions . set ( incomingTx . id , incomingTx ) ; this . _feePriorityQueue . push ( this . _calculateFeePriority ( incomingTx ) , incomingTx . id ) ; return { status : Status . OK } ; } public remove ( tx ) { const foundTx = this . _allTransactions . get ( tx . id ) ; if ( ! foundTx ) { return false ; } this . _allTransactions . delete ( tx . id ) ; debug ( 'Removing from transaction pool with id' , tx . id ) ; const senderId = getAddressFromPublicKey ( foundTx . senderPublicKey ) ; ( this . _transactionList . get ( senderId ) as TransactionList ) . remove ( tx . nonce ) ; if ( ( this . _transactionList . get ( senderId ) as TransactionList ) . size === 0 ) { this . _transactionList . delete ( senderId ) ; } this . _feePriorityQueue . clear ( ) ; for ( const txObject of this . getAll ( ) ) { this . _feePriorityQueue . push ( txObject . feePriority ?? this . _calculateFeePriority ( txObject ) , txObject . id ) ; } return true ; } public getProcessableTransactions ( ) { const processableTransactions = new dataStructures . BufferMap < Transaction [ ] > ( ) ; for ( const list of this . _transactionList . values ( ) ) { const transactions = list . getProcessable ( ) ; if ( transactions . length !== 0 ) { processableTransactions . set ( list . address , [ ... transactions ] ) ; } } return processableTransactions ; } private _calculateFeePriority ( trx ) { return ( trx . fee - this . _calculateMinFee ( trx ) ) / BigInt ( trx . getBytes ( ) . length ) ; } private _calculateMinFee ( trx ) { const foundBaseFee = this . _baseFees . find ( f => f . moduleID === trx . moduleID && f . assetID === trx . assetID ) ; return ( BigInt ( foundBaseFee ?. baseFee ?? BigInt ( 0 ) ) + BigInt ( this . _minFeePerByte * trx . getBytes ( ) . length ) ) ; } private _getStatus ( errorResponse ) { if ( errorResponse . code === ERR_TRANSACTION_VERIFICATION_FAIL && errorResponse . transactionError . code === ERR_NONCE_OUT_OF_BOUNDS_CODE ) { debug ( 'Received UNPROCESSABLE transaction' ) ; return TransactionStatus . UNPROCESSABLE ; } debug ( 'Received INVALID transaction' ) ; return TransactionStatus . INVALID ; } private _evictUnprocessable ( ) { const unprocessableFeePriorityHeap = new dataStructures . MinHeap < Transaction > ( ) ; for ( const txList of this . _transactionList . values ( ) ) { const unprocessableTransactions = txList . getUnprocessable ( ) ; for ( const unprocessableTx of unprocessableTransactions ) { unprocessableFeePriorityHeap . push ( unprocessableTx . feePriority as bigint , unprocessableTx ) ; } } if ( unprocessableFeePriorityHeap . count < 1 ) { return false ; } const evictedTransaction = unprocessableFeePriorityHeap . pop ( ) ; if ( ! evictedTransaction ) { return false ; } this . events . emit ( events . EVENT_TRANSACTION_REMOVED , { id : evictedTransaction . value . id , nonce : evictedTransaction . value . nonce . toString ( ) , senderPublicKey : evictedTransaction . value . senderPublicKey , reason : 'Pool exceeded the size limit' , } ) ; return this . remove ( evictedTransaction . value ) ; } private _evictProcessable ( ) { const processableFeePriorityHeap = new dataStructures . MinHeap < Transaction > ( ) ; for ( const txList of this . _transactionList . values ( ) ) { const processableTransactions = txList . getProcessable ( ) ; if ( processableTransactions . length ) { const processableTransactionWithHighestNonce = processableTransactions [ processableTransactions . length - 1 ] ; processableFeePriorityHeap . push ( processableTransactionWithHighestNonce . feePriority as bigint , processableTransactionWithHighestNonce ) ; } } if ( processableFeePriorityHeap . count < 1 ) { return false ; } const evictedTransaction = processableFeePriorityHeap . pop ( ) ; if ( ! evictedTransaction ) { return false ; } this . events . emit ( events . EVENT_TRANSACTION_REMOVED , { id : evictedTransaction . value . id , nonce : evictedTransaction . value . nonce . toString ( ) , senderPublicKey : evictedTransaction . value . senderPublicKey , reason : 'Pool exceeded the size limit' , } ) ; return this . remove ( evictedTransaction . value ) ; } private async _reorganize ( ) { for ( const txList of this . _transactionList . values ( ) ) { const promotableTransactions = txList . getPromotable ( ) ; if ( ! promotableTransactions . length ) { continue ; } const processableTransactions = txList . getProcessable ( ) ; const allTransactions = [ ... processableTransactions , ... promotableTransactions ] ; let firstInvalidTransaction ; try { await this . _applyFunction ( allTransactions ) ; } catch ( error ) { const failedStatus = this . _getStatus ( error ) ; firstInvalidTransaction = { id : ( error as TransactionFailedResponse ) . id , status : failedStatus , } ; } const successfulTransactionIds = [ ] ; for ( const tx of allTransactions ) { if ( firstInvalidTransaction && tx . id . equals ( firstInvalidTransaction . id ) ) { break ; } successfulTransactionIds . push ( tx . id ) ; } txList . promote ( promotableTransactions . filter ( tx => successfulTransactionIds . includes ( tx . id ) ) ) ; const invalidTransaction = firstInvalidTransaction && firstInvalidTransaction . status === TransactionStatus . INVALID ? allTransactions . find ( tx => tx . id . equals ( firstInvalidTransaction ?. id as Buffer ) ) : undefined ; if ( invalidTransaction ) { for ( const tx of allTransactions ) { if ( tx . nonce >= invalidTransaction . nonce ) { this . events . emit ( events . EVENT_TRANSACTION_REMOVED , { id : tx . id , nonce : tx . nonce . toString ( ) , senderPublicKey : tx . senderPublicKey , reason : `Invalid transaction ${ invalidTransaction . id . toString ( 'binary' ) }` , } ) ; this . remove ( tx ) ; } } } } } private async _expire ( ) { for ( const transaction of this . _allTransactions . values ( ) ) { const timeDifference = Math . round ( Math . abs ( ( transaction . receivedAt as Date ) . getTime ( ) - new Date ( ) . getTime ( ) ) ) ; if ( timeDifference > this . _transactionExpiryTime ) { this . events . emit ( events . EVENT_TRANSACTION_REMOVED , { id : transaction . id , nonce : transaction . nonce . toString ( ) , senderPublicKey : transaction . senderPublicKey , reason : 'Transaction exceeded the expiry time' , } ) ; this . remove ( transaction ) ; } } } }