client -> tx pool: add txs to the pool, log pool statistics

packages/client/lib/service/txpool.ts

@@ -2,14 +2,20 @@ import EventEmitter from 'events'
 import { Config } from '../config'
 import { Peer } from '../net/peer'
 import { EthProtocolMethods } from '../net/protocol'
-import { TransactionFactory } from '@ethereumjs/tx'
+import { TransactionFactory, TypedTransaction } from '@ethereumjs/tx'
 
 export interface TxPoolOptions {
   /* Config */
   config: Config
 }
 
+export type TxPoolObject = {
+  tx: TypedTransaction
+  added: number
+}
+
 const TX_RETRIEVAL_LIMIT = 256
+const LOG_STATISTICS_INTERVAL = 3000 // ms
 
 /**
  * @module service
@@ -30,16 +36,12 @@ export class TxPool extends EventEmitter {
   private pending: string[] = []
 
   /**
-   * List of pooled tx hashes
+   * List of handled tx hashes
+   * (have been added to the pool at some point)
    */
-  private pooled: string[] = []
+  private handled: string[] = []
 
-  /**
-   * List of outdated tx hashes
-   *
-   * Simple FIFO list to avoid double requests on outdated txs
-   */
-  private outdated: string[] = []
+  public pool: Map<string, TxPoolObject[]>
 
   /**
    * Create new tx pool
@@ -50,7 +52,7 @@ export class TxPool extends EventEmitter {
 
     this.config = options.config
 
-    //this.pool = new Map<string, Peer>()
+    this.pool = new Map<string, TxPoolObject[]>()
     this.opened = false
 
     this.init()
@@ -68,6 +70,8 @@ export class TxPool extends EventEmitter {
       return false
     }
     this.opened = true
+
+    setInterval(this._logPoolStats.bind(this), LOG_STATISTICS_INTERVAL)
     return true
   }
 
@@ -78,12 +82,12 @@ export class TxPool extends EventEmitter {
    */
   async announced(txHashes: Buffer[], peer: Peer) {
     if (txHashes.length) {
-      this.config.logger.info(`Tx pool: received new pooled tx hashes number=${txHashes.length}`)
+      //this.config.logger.info(`Tx pool: received new pooled tx hashes number=${txHashes.length}`)
 
       const reqHashes = []
       for (const txHash of txHashes) {
         const txHashStr = txHash.toString('hex')
-        if (txHashStr in this.pending || txHashStr in this.pooled || txHashStr in this.outdated) {
+        if (txHashStr in this.pending || txHashStr in this.handled) {
           continue
         }
         reqHashes.push(txHash)
@@ -102,15 +106,33 @@ export class TxPool extends EventEmitter {
           this.pending = this.pending.filter((hash) => hash !== reqHashStr)
         }
 
-        this.config.logger.info(`Tx pool: received txs number=${txsResult.length}`)
+        //this.config.logger.info(`Tx pool: received txs number=${txsResult.length}`)
         for (const txData of txsResult[1]) {
           const tx = TransactionFactory.fromBlockBodyData(txData)
-          this.config.logger.info(`Nonce: ${tx.nonce}`)
+          const sender = tx.getSenderAddress().toString()
+          const inPool = this.pool.get(sender)
+          let add: TxPoolObject[] = []
+          if (inPool) {
+            // Replace pooled txs with the same nonce
+            add = inPool.filter((poolObj) => poolObj.tx.nonce !== tx.nonce)
+          }
+          add.push({ tx, added: Date.now() })
+
+          this.pool.set(tx.getSenderAddress().toString(), add)
+          this.handled.push(tx.hash().toString('hex'))
         }
       }
     }
   }
 
+  _logPoolStats() {
+    let count = 0
+    this.pool.forEach((poolObjects) => {
+      count += poolObjects.length
+    })
+    this.config.logger.info(`TxPool Statistics transactions=${count} senders=${this.pool.size}`)
+  }
+
   /**
    * Close pool
    */