Switch events to eventemitter3 - the saga (#3746)

* switch events to eventemitter3

* lint

* remove cruft

* make client play nice

* fix tests

* Move blockchain off of asyncEventEmitter

* Modify evm emit function to handle async events

* update vm to use async events

* Allow sync and async listeners

* remove asynceventemitter

* typo and make typescript happy

* typo

* add events benchmark

* address feedback

* Call main function in example outside of own definition

---------

Co-authored-by: Amir <indigophi@protonmail.com>

packages/blockchain/package.json

@@ -49,10 +49,11 @@
   "dependencies": {
     "@ethereumjs/block": "^5.3.0",
     "@ethereumjs/common": "^4.4.0",
-    "@ethereumjs/rlp": "^5.0.2",
     "@ethereumjs/mpt": "^6.2.2",
+    "@ethereumjs/rlp": "^5.0.2",
     "@ethereumjs/util": "^9.1.0",
     "debug": "^4.3.3",
+    "eventemitter3": "^5.0.1",
     "lru-cache": "10.1.0"
   },
   "devDependencies": {

packages/blockchain/src/blockchain.ts

@@ -1,7 +1,6 @@
 import { Block, BlockHeader, createBlock } from '@ethereumjs/block'
 import { Common, ConsensusAlgorithm, ConsensusType, Hardfork, Mainnet } from '@ethereumjs/common'
 import {
-  AsyncEventEmitter,
   BIGINT_0,
   BIGINT_1,
   BIGINT_8,
@@ -15,6 +14,7 @@ import {
   equalsBytes,
 } from '@ethereumjs/util'
 import debugDefault from 'debug'
+import { EventEmitter } from 'eventemitter3'
 
 import { CasperConsensus } from './consensus/casper.js'
 import {
@@ -28,7 +28,7 @@ import { DBManager } from './db/manager.js'
 import { DBTarget } from './db/operation.js'
 
 import type {
-  BlockchainEvents,
+  BlockchainEvent,
   BlockchainInterface,
   BlockchainOptions,
   Consensus,
@@ -53,7 +53,7 @@ import type { Debugger } from 'debug'
 export class Blockchain implements BlockchainInterface {
   db: DB<Uint8Array | string, Uint8Array | string | DBObject>
   dbManager: DBManager
-  events: AsyncEventEmitter<BlockchainEvents>
+  events: EventEmitter<BlockchainEvent>
 
   private _genesisBlock?: Block /** The genesis block of this blockchain */
   private _customGenesisState?: GenesisState /** Custom genesis state */
@@ -129,7 +129,7 @@ export class Blockchain implements BlockchainInterface {
 
     this.dbManager = new DBManager(this.db, this.common)
 
-    this.events = new AsyncEventEmitter()
+    this.events = new EventEmitter<BlockchainEvent>()
 
     this._consensusDict = {}
     this._consensusDict[ConsensusAlgorithm.Casper] = new CasperConsensus()

packages/blockchain/src/types.ts

@@ -1,11 +1,12 @@
 import type { Blockchain } from './index.js'
 import type { Block, BlockHeader } from '@ethereumjs/block'
 import type { Common, ConsensusAlgorithm } from '@ethereumjs/common'
-import type { AsyncEventEmitter, DB, DBObject, GenesisState } from '@ethereumjs/util'
+import type { DB, DBObject, GenesisState } from '@ethereumjs/util'
+import type { EventEmitter } from 'eventemitter3'
 
 export type OnBlock = (block: Block, reorg: boolean) => Promise<void> | void
 
-export type BlockchainEvents = {
+export type BlockchainEvent = {
   deletedCanonicalBlocks: (data: Block[], resolve?: (result?: any) => void) => void
 }
 
@@ -87,7 +88,7 @@ export interface BlockchainInterface {
   /**
    * Optional events emitter
    */
-  events?: AsyncEventEmitter<BlockchainEvents>
+  events?: EventEmitter<BlockchainEvent>
 }
 
 export interface GenesisOptions {

packages/client/bin/cli.ts

@@ -807,6 +807,7 @@ async function inputAccounts() {
     // @ts-ignore Looks like there is a type incompatibility in NodeJS ReadStream vs what this package expects
     // TODO: See whether package needs to be updated or not
     input: process.stdin,
+    // @ts-ignore
     output: process.stdout,
   })
 
@@ -1179,7 +1180,7 @@ async function run() {
     ignoreStatelessInvalidExecs: args.ignoreStatelessInvalidExecs,
     prometheusMetrics,
   })
-  config.events.setMaxListeners(50)
+
   config.events.on(Event.SERVER_LISTENING, (details) => {
     const networkDir = config.getNetworkDirectory()
     // Write the transport into a file

packages/client/package.json

@@ -62,14 +62,15 @@
     "@ethereumjs/ethash": "3.0.4",
     "@ethereumjs/evm": "3.1.0",
     "@ethereumjs/genesis": "0.2.3",
+    "@ethereumjs/mpt": "6.2.2",
     "@ethereumjs/rlp": "5.0.2",
     "@ethereumjs/statemanager": "2.4.0",
-    "@ethereumjs/mpt": "6.2.2",
     "@ethereumjs/tx": "5.4.0",
     "@ethereumjs/util": "9.1.0",
     "@ethereumjs/vm": "8.1.0",
     "@js-sdsl/ordered-map": "^4.4.2",
     "@multiformats/multiaddr": "^12.2.1",
+    "@paulmillr/trusted-setups": "^0.1.2",
     "@polkadot/wasm-crypto": "^7.3.2",
     "@scure/base": "^1.1.7",
     "abstract-level": "^1.0.3",
@@ -79,12 +80,12 @@
     "cors": "^2.8.5",
     "debug": "^4.3.3",
     "ethereum-cryptography": "^3.0.0",
+    "eventemitter3": "^5.0.1",
     "jayson": "^4.0.0",
-    "@paulmillr/trusted-setups": "^0.1.2",
-    "micro-eth-signer": "^0.11.0",
     "level": "^8.0.0",
     "mcl-wasm": "^1.5.0",
     "memory-level": "^1.0.0",
+    "micro-eth-signer": "^0.11.0",
     "prom-client": "^15.1.0",
     "rustbn-wasm": "^0.4.0",
     "verkle-cryptography-wasm": "^0.4.8",

packages/client/src/config.ts

@@ -1,15 +1,16 @@
 import { Common, Hardfork, Mainnet } from '@ethereumjs/common'
 import { genPrivateKey } from '@ethereumjs/devp2p'
 import { type Address, BIGINT_0, BIGINT_1, BIGINT_2, BIGINT_256 } from '@ethereumjs/util'
+import { EventEmitter } from 'eventemitter3'
 import { Level } from 'level'
 
 import { getLogger } from './logging.js'
 import { RlpxServer } from './net/server/index.js'
-import { Event, EventBus } from './types.js'
+import { Event } from './types.js'
 import { isBrowser, short } from './util/index.js'
 
 import type { Logger } from './logging.js'
-import type { EventBusType, MultiaddrLike, PrometheusMetrics } from './types.js'
+import type { EventParams, MultiaddrLike, PrometheusMetrics } from './types.js'
 import type { BlockHeader } from '@ethereumjs/block'
 import type { VM, VMProfilerOpts } from '@ethereumjs/vm'
 import type { Multiaddr } from '@multiformats/multiaddr'
@@ -355,7 +356,7 @@ export class Config {
    * Central event bus for events emitted by the different
    * components of the client
    */
-  public readonly events: EventBusType
+  public readonly events: EventEmitter<EventParams>
 
   public static readonly CHAIN_DEFAULT = Mainnet
   public static readonly SYNCMODE_DEFAULT = SyncMode.Full
@@ -478,7 +479,7 @@ export class Config {
   public readonly metrics: PrometheusMetrics | undefined
 
   constructor(options: ConfigOptions = {}) {
-    this.events = new EventBus() as EventBusType
+    this.events = new EventEmitter<EventParams>()
 
     this.syncmode = options.syncmode ?? Config.SYNCMODE_DEFAULT
     this.vm = options.vm

packages/client/src/net/peer/peer.ts

@@ -1,5 +1,5 @@
 import { BIGINT_0, short } from '@ethereumjs/util'
-import { EventEmitter } from 'events'
+import { EventEmitter } from 'eventemitter3'
 
 import { BoundEthProtocol, BoundLesProtocol, BoundSnapProtocol } from '../protocol/index.js'
 

packages/client/src/net/protocol/sender.ts

@@ -1,4 +1,4 @@
-import { EventEmitter } from 'events'
+import { EventEmitter } from 'eventemitter3'
 
 /**
  * Base class for transport specific message sender/receiver. Subclasses should

packages/client/src/sync/fullsync.ts

@@ -70,8 +70,8 @@ export class FullSynchronizer extends Synchronizer {
     const syncEvent: Promise<boolean> = new Promise((resolve) => {
       // This event listener listens for other instances of the fetcher that might be syncing from a different peer
       // and reach the head of the chain before the current fetcher.
-      this.config.events.once(Event.SYNC_SYNCHRONIZED, (height?: number) => {
-        this.resolveSync(height)
+      this.config.events.once(Event.SYNC_SYNCHRONIZED, (chainHeight?: bigint) => {
+        this.resolveSync(chainHeight)
         resolve(true)
       })
     })

packages/client/src/sync/sync.ts

@@ -145,9 +145,9 @@ export abstract class Synchronizer {
 
   abstract syncWithPeer(peer?: Peer): Promise<boolean>
 
-  resolveSync(height?: number) {
+  resolveSync(height?: bigint) {
     this.clearFetcher()
-    const heightStr = typeof height === 'number' && height !== 0 ? ` height=${height}` : ''
+    const heightStr = typeof height === 'bigint' && height !== BIGINT_0 ? ` height=${height}` : ''
     this.config.logger.debug(`Finishing up sync with the current fetcher ${heightStr}`)
     return true
   }

packages/client/src/types.ts

@@ -1,5 +1,3 @@
-import { EventEmitter } from 'events'
-
 import type { SyncMode } from './index.js'
 import type { Peer } from './net/peer/index.js'
 import type { Server } from './net/server/index.js'
@@ -56,33 +54,6 @@ export interface EventParams {
   [Event.PROTOCOL_MESSAGE]: [messageDetails: any, protocolName: string, sendingPeer: Peer]
 }
 
-export declare interface EventBus<T extends Event> {
-  emit(event: T, ...args: EventParams[T]): boolean
-  on(event: T, listener: (...args: EventParams[T]) => void): this
-}
-
-// eslint-disable-next-line @typescript-eslint/no-unused-vars
-export class EventBus<T extends Event> extends EventEmitter {}
-export type EventBusType = EventBus<Event.CHAIN_UPDATED> &
-  EventBus<Event.CLIENT_SHUTDOWN> &
-  EventBus<Event.SYNC_EXECUTION_VM_ERROR> &
-  EventBus<Event.SYNC_FETCHED_BLOCKS> &
-  EventBus<Event.SYNC_FETCHED_HEADERS> &
-  EventBus<Event.SYNC_SYNCHRONIZED> &
-  EventBus<Event.SYNC_SNAPSYNC_COMPLETE> &
-  EventBus<Event.SYNC_FETCHER_ERROR> &
-  EventBus<Event.PEER_CONNECTED> &
-  EventBus<Event.PEER_DISCONNECTED> &
-  EventBus<Event.PEER_ERROR> &
-  EventBus<Event.SERVER_LISTENING> &
-  EventBus<Event.SERVER_ERROR> &
-  EventBus<Event.SYNC_ERROR> &
-  EventBus<Event.POOL_PEER_ADDED> &
-  EventBus<Event.POOL_PEER_REMOVED> &
-  EventBus<Event.POOL_PEER_BANNED> &
-  EventBus<Event.PROTOCOL_ERROR> &
-  EventBus<Event.PROTOCOL_MESSAGE>
-
 /**
  * Like types
  */

packages/client/test/integration/mocks/mockpeer.ts

@@ -1,4 +1,4 @@
-import { EventEmitter } from 'events'
+import { EventEmitter } from 'eventemitter3'
 import { pipe } from 'it-pipe'
 import pushable from 'it-pushable'
 

packages/client/test/integration/mocks/mocksender.ts

@@ -1,7 +1,7 @@
 import { Sender } from '../../../src/net/protocol/index.js'
 
 import type { Pushable } from './mockpeer.js'
-import type EventEmitter from 'events'
+import type { EventEmitter } from 'eventemitter3'
 
 export class MockSender extends Sender {
   public protocol: string

packages/client/test/integration/mocks/network.ts

@@ -1,4 +1,4 @@
-import { EventEmitter } from 'events'
+import { EventEmitter } from 'eventemitter3'
 //@ts-ignore
 import DuplexPair from 'it-pair/duplex'
 

packages/client/test/miner/miner.spec.ts

@@ -140,26 +140,24 @@ const customCommon = createCommonFromGethGenesis(chainData, {
   chain: 'devnet',
   hardfork: Hardfork.Berlin,
 })
-customCommon.events.setMaxListeners(50)
+
 const customConfig = new Config({
   accountCache: 10000,
   storageCache: 1000,
   accounts,
   mine: true,
   common: customCommon,
 })
-customConfig.events.setMaxListeners(50)
 
 const goerliCommon = new Common({ chain: Goerli, hardfork: Hardfork.Berlin })
-goerliCommon.events.setMaxListeners(50)
+
 const goerliConfig = new Config({
   accountCache: 10000,
   storageCache: 1000,
   accounts,
   mine: true,
   common: customCommon,
 })
-customConfig.events.setMaxListeners(50)
 
 const createTx = (
   from = A,

packages/client/test/net/peer/rlpxpeer.spec.ts

@@ -1,4 +1,4 @@
-import { EventEmitter } from 'events'
+import { EventEmitter } from 'eventemitter3'
 import { assert, describe, expect, it, vi } from 'vitest'
 
 import { Config } from '../../../src/config.js'

packages/client/test/net/peerpool.spec.ts

@@ -1,4 +1,4 @@
-import { EventEmitter } from 'events'
+import { EventEmitter } from 'eventemitter3'
 import { assert, describe, it, vi } from 'vitest'
 
 import { Config } from '../../src/config.js'

packages/client/test/net/protocol/boundprotocol.spec.ts

@@ -1,5 +1,5 @@
 /// <reference path="./testdouble.d.ts" />
-import { EventEmitter } from 'events'
+import { EventEmitter } from 'eventemitter3'
 import * as td from 'testdouble'
 import { assert, describe, it } from 'vitest'