Client: add basic TxPool

packages/client/lib/net/protocol/ethprotocol.ts

@@ -29,13 +29,21 @@ type GetBlockBodiesOpts = {
   hashes: Buffer[]
 }
 
+type GetPooledTransactionsOpts = {
+  /* Request id (default: next internal id) */
+  reqId?: BN
+  /* The tx hashes */
+  hashes: Buffer[]
+}
+
 /*
  * Messages with responses that are added as
  * methods in camelCase to BoundProtocol.
  */
 export interface EthProtocolMethods {
   getBlockHeaders: (opts: GetBlockHeadersOpts) => Promise<[BN, BlockHeader[]]>
   getBlockBodies: (opts: GetBlockBodiesOpts) => Promise<[BN, BlockBodyBuffer[]]>
+  getPooledTransactions: (opts: GetPooledTransactionsOpts) => Promise<[BN, any[]]>
 }
 
 const id = new BN(0)
@@ -113,6 +121,24 @@ export class EthProtocol extends Protocol {
       ],
       decode: ([reqId, bodies]: [Buffer, BlockBodyBuffer[]]) => [new BN(reqId), bodies],
     },
+    {
+      name: 'NewPooledTransactionHashes',
+      code: 0x08,
+    },
+    {
+      name: 'GetPooledTransactions',
+      code: 0x09,
+      response: 0x0a,
+      encode: ({ reqId, hashes }: GetPooledTransactionsOpts) => [
+        (reqId === undefined ? id.iaddn(1) : new BN(reqId)).toArrayLike(Buffer),
+        hashes,
+      ],
+    },
+    {
+      name: 'PooledTransactions',
+      code: 0x0a,
+      decode: ([reqId, txs]: [Buffer, any[]]) => [new BN(reqId), txs],
+    },
   ]
 
   /**

packages/client/lib/service/ethereumservice.ts

@@ -92,4 +92,15 @@ export class EthereumService extends Service {
     await this.synchronizer.stop()
     await super.stop()
   }
+
+  /**
+   * Close service.
+   * @return {Promise}
+   */
+  async close() {
+    if (this.opened) {
+      await this.synchronizer.close()
+    }
+    await super.close()
+  }
 }

packages/client/lib/service/fullethereumservice.ts

@@ -94,6 +94,8 @@ export class FullEthereumService extends EthereumService {
       peer.eth!.send('BlockBodies', { reqId, bodies })
     } else if (message.name === 'NewBlockHashes') {
       this.synchronizer.handleNewBlockHashes(message.data)
+    } else if (message.name === 'NewPooledTransactionHashes') {
+      await this.synchronizer.txPool.announcedTxHashes(message.data, peer)
     }
   }
 

packages/client/lib/service/service.ts

@@ -97,7 +97,7 @@ export class Service {
    * @return {Promise}
    */
   async close() {
-    if (this.running) {
+    if (this.opened) {
       await this.pool.close()
     }
     this.opened = false

packages/client/lib/sync/fullsync.ts

@@ -5,6 +5,7 @@ import { Synchronizer, SynchronizerOptions } from './sync'
 import { BlockFetcher } from './fetcher'
 import { Block } from '@ethereumjs/block'
 import { VMExecution } from './execution/vmexecution'
+import { TxPool } from './txpool'
 import { Event } from '../types'
 
 /**
@@ -14,6 +15,8 @@ import { Event } from '../types'
 export class FullSynchronizer extends Synchronizer {
   public execution: VMExecution
 
+  public txPool: TxPool
+
   constructor(options: SynchronizerOptions) {
     super(options)
 
@@ -23,13 +26,18 @@ export class FullSynchronizer extends Synchronizer {
       chain: options.chain,
     })
 
+    this.txPool = new TxPool({
+      config: this.config,
+    })
+
     this.config.events.on(Event.SYNC_EXECUTION_VM_ERROR, async () => {
       await this.stop()
     })
 
     this.config.events.on(Event.CHAIN_UPDATED, async () => {
       if (this.running) {
         await this.execution.run()
+        this.checkTxPoolState()
       }
     })
     // eslint-disable-next-line @typescript-eslint/no-floating-promises
@@ -44,6 +52,28 @@ export class FullSynchronizer extends Synchronizer {
     return 'full'
   }
 
+  /**
+   * Open synchronizer. Must be called before sync() is called
+   */
+  async open(): Promise<void> {
+    await super.open()
+    await this.chain.open()
+    await this.execution.open()
+    this.txPool.open()
+    await this.pool.open()
+    this.execution.syncing = true
+    const number = this.chain.blocks.height.toNumber()
+    const td = this.chain.blocks.td.toString(10)
+    const hash = this.chain.blocks.latest!.hash()
+    this.startingBlock = number
+    this.config.chainCommon.setHardforkByBlockNumber(number)
+    this.config.logger.info(
+      `Latest local block: number=${number} td=${td} hash=${short(
+        hash
+      )} hardfork=${this.config.chainCommon.hardfork()}`
+    )
+  }
+
   /**
    * Returns true if peer can be used for syncing
    * @return {boolean}
@@ -86,6 +116,21 @@ export class FullSynchronizer extends Synchronizer {
     return result ? result[1][0] : undefined
   }
 
+  /**
+   * Checks if tx pool should be started
+   */
+  checkTxPoolState() {
+    if (!this.syncTargetHeight || this.txPool.running) {
+      return
+    }
+    // If height gte target, we are close enough to the
+    // head of the chain that the tx pool can be started
+    const target = this.syncTargetHeight.subn(this.txPool.BLOCKS_BEFORE_TARGET_HEIGHT_ACTIVATION)
+    if (this.chain.headers.height.gte(target)) {
+      this.txPool.start()
+    }
+  }
+
   /**
    * Sync all blocks and state from peer starting from current height.
    * @param  peer remote peer to sync with
@@ -139,6 +184,7 @@ export class FullSynchronizer extends Synchronizer {
             this.pool.size
           }`
         )
+        this.txPool.newBlocks(blocks)
       })
 
       this.config.events.on(Event.SYNC_SYNCHRONIZED, () => {
@@ -153,33 +199,14 @@ export class FullSynchronizer extends Synchronizer {
     })
   }
 
-  /**
-   * Open synchronizer. Must be called before sync() is called
-   */
-  async open(): Promise<void> {
-    await this.chain.open()
-    await this.execution.open()
-    await this.pool.open()
-    this.execution.syncing = true
-    const number = this.chain.blocks.height.toNumber()
-    const td = this.chain.blocks.td.toString(10)
-    const hash = this.chain.blocks.latest!.hash()
-    this.startingBlock = number
-    this.config.chainCommon.setHardforkByBlockNumber(number)
-    this.config.logger.info(
-      `Latest local block: number=${number} td=${td} hash=${short(
-        hash
-      )} hardfork=${this.config.chainCommon.hardfork()}`
-    )
-  }
-
   /**
    * Stop synchronization. Returns a promise that resolves once its stopped.
    * @return {Promise}
    */
   async stop(): Promise<boolean> {
     this.execution.syncing = false
     await this.execution.stop()
+    this.txPool.stop()
 
     if (!this.running) {
       return false
@@ -195,4 +222,15 @@ export class FullSynchronizer extends Synchronizer {
 
     return true
   }
+
+  /**
+   * Close synchronizer.
+   * @return {Promise}
+   */
+  async close() {
+    if (this.opened) {
+      this.txPool.close()
+    }
+    await super.close()
+  }
 }

packages/client/lib/sync/lightsync.ts

@@ -23,6 +23,20 @@ export class LightSynchronizer extends Synchronizer {
     return 'light'
   }
 
+  /**
+   * Open synchronizer. Must be called before sync() is called
+   */
+  async open(): Promise<void> {
+    await super.open()
+    await this.chain.open()
+    await this.pool.open()
+    const number = this.chain.headers.height.toNumber()
+    const td = this.chain.headers.td.toString(10)
+    const hash = this.chain.blocks.latest!.hash()
+    this.startingBlock = number
+    this.config.logger.info(`Latest local header: number=${number} td=${td} hash=${short(hash)}`)
+  }
+
   /**
    * Returns true if peer can be used for syncing
    * @return {boolean}
@@ -131,19 +145,6 @@ export class LightSynchronizer extends Synchronizer {
     })
   }
 
-  /**
-   * Open synchronizer. Must be called before sync() is called
-   */
-  async open(): Promise<void> {
-    await this.chain.open()
-    await this.pool.open()
-    const number = this.chain.headers.height.toNumber()
-    const td = this.chain.headers.td.toString(10)
-    const hash = this.chain.blocks.latest!.hash()
-    this.startingBlock = number
-    this.config.logger.info(`Latest local header: number=${number} td=${td} hash=${short(hash)}`)
-  }
-
   /**
    * Stop synchronization. Returns a promise that resolves once its stopped.
    * @return {Promise}