Move all event handling to PortalNetwork class (#793)

* update event types

* remove EventEmitter inheritance from Network classes

* update tests

Author: ScottyPoi

packages/portalnetwork/src/client/types.ts

@@ -17,10 +17,27 @@ export interface INodeAddress {
   nodeId: NodeId
 }
 
-export interface PortalNetworkEvents {
-  NodeAdded: (nodeId: NodeId, networkId: NetworkId) => void
-  NodeRemoved: (nodeId: NodeId, networkId: NetworkId) => void
-  ContentAdded: (key: Uint8Array, contentType: number, content: string) => void
+type ContentAddedEventName = `${NetworkId}:ContentAdded`
+type ContentAddedEventType = (key: Uint8Array, content: Uint8Array) => Promise<void | { content: Uint8Array; utp: boolean }>
+type ContentAddedEvents = {
+  [K in ContentAddedEventName]: ContentAddedEventType
+}
+
+type NodeAddedEventName = `${NetworkId}:NodeAdded`
+type NodeAddedEventType = (nodeId: NodeId) => void
+type NodeAddedEvents = {
+  [K in NodeAddedEventName]: NodeAddedEventType
+}
+
+type NodeRemovedEventName = `${NetworkId}:NodeRemoved`
+type NodeRemovedEventType = (nodeId: NodeId) => void
+type NodeRemovedEvents = {
+  [K in NodeRemovedEventName]: NodeRemovedEventType
+}
+
+type NetworkEvents = ContentAddedEvents & NodeAddedEvents & NodeRemovedEvents
+
+export interface PortalNetworkEvents extends NetworkEvents {
   Verified: (key: Uint8Array, verified: boolean) => void
   SendTalkReq: (nodeId: string, requestId: string, payload: string) => void
   SendTalkResp: (nodeId: string, requestId: string, payload: string) => void

packages/portalnetwork/src/networks/beacon/beacon.ts

@@ -90,7 +90,7 @@ export class BeaconNetwork extends BaseNetwork {
     this.logger = debug(this.enr.nodeId.slice(0, 5)).extend('Portal').extend('BeaconNetwork')
     this.routingTable.setLogger(this.logger)
     this.forkDigest = Uint8Array.from([0, 0, 0, 0])
-    this.on('ContentAdded', async (contentKey: Uint8Array) => {
+    this.portal.on(`${this.networkId}:ContentAdded`, async (contentKey: Uint8Array) => {
       if (contentKey[0] === BeaconNetworkContentType.LightClientUpdate) {
         // don't gossip individual LightClientUpdates since they aren't officially supported
         return
@@ -111,14 +111,14 @@ export class BeaconNetwork extends BaseNetwork {
     switch (this.syncStrategy) {
       case SyncStrategy.PollNetwork:
         this.bootstrapFinder = new Map()
-        this.portal.on('NodeAdded', this.getBootStrapVote)
+        this.portal.on(`${this.networkId}:NodeAdded`, this.getBootStrapVote)
         break
       case SyncStrategy.TrustedBlockRoot:
         if (trustedBlockRoot === undefined)
           throw new Error('must provided trusted block root with SyncStrategy.TrustedBlockRoot')
         this.bootstrapFinder = new Map()
         this.trustedBlockRoot = trustedBlockRoot
-        this.portal.on('NodeAdded', this.getBootstrap)
+        this.portal.on(`${this.networkId}:NodeAdded`, this.getBootstrap)
         break
     }
   }
@@ -130,9 +130,7 @@ export class BeaconNetwork extends BaseNetwork {
    * @param nodeId NodeId for a peer that was just discovered by the Portal Network `client`
    * @param network the network ID for the node just discovered
    */
-  private getBootstrap = async (nodeId: string, network: NetworkId) => {
-    // We check the network ID because NodeAdded is emitted regardless of network
-    if (network !== this.networkId) return
+  private getBootstrap = async (nodeId: string) => {
     const enr = getENR(this.routingTable, nodeId)
     if (enr === undefined) return
     const decoded = await this.sendFindContent(
@@ -153,7 +151,7 @@ export class BeaconNetwork extends BaseNetwork {
       )
       if (headerHash === this.trustedBlockRoot) {
         void this.initializeLightClient(headerHash)
-        this.portal.removeListener('NodeAdded', this.getBootstrap)
+        this.portal.removeListener(`${this.networkId}:NodeAdded`, this.getBootstrap)
       }
     }
   }
@@ -168,9 +166,9 @@ export class BeaconNetwork extends BaseNetwork {
    * @param nodeId NodeId for a peer that was just discovered by the Portal Network `client`
    * @param network the network ID for the node just discovered
    */
-  private getBootStrapVote = async (nodeId: string, network: NetworkId) => {
+  private getBootStrapVote = async (nodeId: string) => {
     try {
-      if (network === this.networkId) {
+      
         // We check the network ID because NodeAdded is emitted regardless of network
         if (this.bootstrapFinder.has(nodeId)) {
           return
@@ -257,7 +255,7 @@ export class BeaconNetwork extends BaseNetwork {
                   ssz[fork].LightClientBootstrap.deserialize(res.content.slice(4))
                   this.logger.extend('BOOTSTRAP')(`found a valid bootstrap - ${results[x][0]}`)
                   await this.store(bootstrapKey, res.content)
-                  this.portal.removeListener('NodeAdded', this.getBootStrapVote)
+                  this.portal.removeListener(`${this.networkId}:NodeAdded`, this.getBootStrapVote)
                   this.logger.extend('BOOTSTRAP')('Terminating Light Client bootstrap process')
                   await this.initializeLightClient(results[x][0])
                   return
@@ -274,7 +272,7 @@ export class BeaconNetwork extends BaseNetwork {
             this.bootstrapFinder.set(peer, {})
           }
         }
-      }
+      
     } catch (err) {
       this.logger.extend('BOOTSTRAP')(err)
     }
@@ -287,8 +285,8 @@ export class BeaconNetwork extends BaseNetwork {
    */
   public initializeLightClient = async (blockRoot: string) => {
     // Ensure bootstrap finder mechanism is disabled if currently running
-    this.portal.removeListener('NodeAdded', this.getBootStrapVote)
-    this.portal.removeListener('NodeAdded', this.getBootstrap)
+    this.portal.removeListener(`${this.networkId}:NodeAdded`, this.getBootStrapVote)
+    this.portal.removeListener(`${this.networkId}:NodeAdded`, this.getBootstrap)
 
     // Setup the Lodestar light client logger using our debug logger
     const lcLogger = this.logger.extend('LightClient')
@@ -473,7 +471,7 @@ export class BeaconNetwork extends BaseNetwork {
             this.logger.extend('FOUNDCONTENT')(`received uTP Connection ID ${id}`)
             response = await new Promise((resolve, _reject) => {
               // TODO: Figure out how to clear this listener
-              this.on('ContentAdded', (contentKey: Uint8Array, value) => {
+              this.portal.on(`${this.networkId}:ContentAdded`, (contentKey: Uint8Array, value) => {
                 if (equalsBytes(contentKey, key) === true) {
                   this.logger.extend('FOUNDCONTENT')(`received content for uTP Connection ID ${id}`)
                   resolve({ content: value, utp: true })
@@ -739,7 +737,7 @@ export class BeaconNetwork extends BaseNetwork {
     this.logger(
       `storing ${BeaconNetworkContentType[contentType]} content corresponding to ${bytesToHex(contentKey)}`,
     )
-    this.emit('ContentAdded', contentKey, value)
+    this.portal.emit(`${this.networkId}:ContentAdded`, contentKey, value)
   }
 
   /**

packages/portalnetwork/src/networks/beacon/ultralightTransport.ts

@@ -227,7 +227,7 @@ export class UltralightTransport implements LightClientTransport {
   }
 
   onOptimisticUpdate(handler: (optimisticUpdate: LightClientOptimisticUpdate) => void): void {
-    this.network.on('ContentAdded', (contentKey: Uint8Array, content: Uint8Array) => {
+    this.network.portal.on(`${this.network.networkId}:ContentAdded`, (contentKey: Uint8Array, content: Uint8Array) => {
       const contentType = contentKey[0]
       if (contentType === BeaconNetworkContentType.LightClientOptimisticUpdate) {
         const forkhash = content.slice(0, 4)
@@ -240,11 +240,14 @@ export class UltralightTransport implements LightClientTransport {
           this.logger('something went wrong trying to process Optimistic Update')
           this.logger(err)
         }
-      }
-    })
+        }
+      },
+    )
   }
   onFinalityUpdate(handler: (finalityUpdate: LightClientFinalityUpdate) => void): void {
-    this.network.on('ContentAdded', (contentKey: Uint8Array, content: Uint8Array) => {
+    this.network.portal.on(
+      `${this.network.networkId}:ContentAdded`,
+      (contentKey: Uint8Array, content: Uint8Array) => {
       const contentType = contentKey[0]
       if (contentType === BeaconNetworkContentType.LightClientFinalityUpdate) {
         const forkhash = content.slice(0, 4)

packages/portalnetwork/src/networks/history/history.ts

@@ -355,7 +355,7 @@ export class HistoryNetwork extends BaseNetwork {
             this.logger.extend('FOUNDCONTENT')(`received uTP Connection ID ${id}`)
             response = await new Promise((resolve, _reject) => {
               // TODO: Figure out how to clear this listener
-              this.on('ContentAdded', (contentKey: Uint8Array, value) => {
+              this.portal.on(`${this.networkId}:ContentAdded`, (contentKey: Uint8Array, value: Uint8Array) => {
                 if (equalsBytes(contentKey, key) === true) {
                   this.logger.extend('FOUNDCONTENT')(`received content for uTP Connection ID ${id}`)
                   resolve({ content: value, utp: true })
@@ -646,7 +646,7 @@ export class HistoryNetwork extends BaseNetwork {
       }
     }
 
-    this.emit('ContentAdded', contentKey, value)
+    this.portal.emit(`${this.networkId}:ContentAdded`, contentKey, value)
     if (this.routingTable.values().length > 0) {
       if (
         contentType !== HistoryNetworkContentType.EphemeralHeader &&
@@ -692,7 +692,7 @@ export class HistoryNetwork extends BaseNetwork {
     const bodyContentKey = getContentKey(HistoryNetworkContentType.BlockBody, hashKey)
     if (block instanceof Block) {
       await this.put(bodyContentKey, bytesToHex(bodyBytes))
-      this.emit('ContentAdded', bodyContentKey, bodyBytes)
+      this.portal.emit(`${this.networkId}:ContentAdded`, bodyContentKey, bodyBytes)
 
       // TODO: Decide when and if to build and store receipts.
       //       Doing this here caused a bottleneck when same receipt is gossiped via uTP at the same time.
@@ -703,7 +703,7 @@ export class HistoryNetwork extends BaseNetwork {
       this.logger('Could not verify block content')
       this.logger('Adding anyway for testing...')
       await this.put(bodyContentKey, bytesToHex(bodyBytes))
-      this.emit('ContentAdded', bodyContentKey, bodyBytes)
+      this.portal.emit(`${this.networkId}:ContentAdded`, bodyContentKey, bodyBytes)
       // TODO: Decide what to do here.  We shouldn't be storing block bodies without a corresponding header
       // as it's against spec
       return

packages/portalnetwork/src/networks/network.ts

@@ -12,7 +12,6 @@ import {
   hexToBytes,
   randomBytes,
 } from '@ethereumjs/util'
-import { EventEmitter } from 'eventemitter3'
 
 import type { ITalkReqMessage } from '@chainsafe/discv5/message'
 import type { SignableENR } from '@chainsafe/enr'
@@ -61,7 +60,8 @@ import { PingPongPayloadExtensions } from '../wire/payloadExtensions.js'
 import { AcceptCode, FoundContent } from '../wire/types.js'
 import { GossipManager } from './gossip.js'
 import { NetworkDB } from './networkDB.js'
-export abstract class BaseNetwork extends EventEmitter {
+
+export abstract class BaseNetwork {
   public capabilities: number[] = [
     PingPongPayloadExtensions.CLIENT_INFO_RADIUS_AND_CAPABILITIES,
     PingPongPayloadExtensions.BASIC_RADIUS_PAYLOAD,
@@ -94,7 +94,6 @@ export abstract class BaseNetwork extends EventEmitter {
     gossipCount,
     dbSize,
   }: BaseNetworkConfig) {
-    super()
     this.bridge = bridge ?? false
     this.networkId = networkId
     this.logger = client.logger.extend(this.constructor.name)
@@ -118,7 +117,7 @@ export abstract class BaseNetwork extends EventEmitter {
       }
     }
     this.gossipManager = new GossipManager(this, gossipCount)
-    this.on('ContentAdded', () => {
+    this.portal.on(`${this.networkId}:ContentAdded`, () => {
       if (this.db.approximateSize / MEGABYTE > this.maxStorage) {
         this.logger(
           `Pruning due to size limit ${this.db.approximateSize / MEGABYTE} > ${this.maxStorage}`,
@@ -1045,14 +1044,14 @@ export abstract class BaseNetwork extends EventEmitter {
         this.logger.extend('RoutingTable')(`adding ${shortId(nodeId)}`)
       this.routingTable.insertOrUpdate(enr, EntryStatus.Connected)
 
-      this.portal.emit('NodeAdded', enr.nodeId, this.networkId)
+      this.portal.emit(`${this.networkId}:NodeAdded`, enr.nodeId)
     } catch (err) {
       this.logger(`Something went wrong: ${(err as any).message}`)
       try {
         this.routingTable.getWithPending(enr as any)?.value === undefined &&
           this.logger(`adding ${enr as any} to ${this.networkName} routing table`)
         this.routingTable.insertOrUpdate(enr, EntryStatus.Connected)
-        this.portal.emit('NodeAdded', enr.nodeId, this.networkId)
+        this.portal.emit(`${this.networkId}:NodeAdded`, enr.nodeId)
       } catch (e) {
         this.logger(`Something went wrong : ${(e as any).message}`)
       }

packages/portalnetwork/src/networks/state/state.ts

@@ -111,7 +111,7 @@ export class StateNetwork extends BaseNetwork {
             this.logger.extend('FOUNDCONTENT')(`received uTP Connection ID ${id}`)
             response = await new Promise((resolve, _reject) => {
               // TODO: Figure out how to clear this listener
-              this.on('ContentAdded', (contentKey: Uint8Array, value) => {
+              this.portal.on(`${this.networkId}:ContentAdded`, (contentKey, value) => {
                 if (equalsBytes(contentKey, key) === true) {
                   this.logger.extend('FOUNDCONTENT')(`received content for uTP Connection ID ${id}`)
                   resolve({ content: value, utp: true })
@@ -192,7 +192,7 @@ export class StateNetwork extends BaseNetwork {
         await this.db.put(contentKey, content)
       }
       this.logger(`content added for: ${bytesToHex(contentKey)}`)
-      this.emit('ContentAdded', contentKey, content)
+      this.portal.emit(`${this.networkId}:ContentAdded`, contentKey, content)
       this.gossipManager.add(contentKey)
     } catch (err: any) {
       this.logger(`Error storing content: ${err.message}`)
@@ -257,7 +257,7 @@ export class StateNetwork extends BaseNetwork {
     }
     for (const { contentKey, dbContent } of interested) {
       await this.db.put(contentKey, dbContent)
-      this.emit('ContentAdded', contentKey, dbContent)
+      this.portal.emit(`${this.networkId}:ContentAdded`, contentKey, dbContent)
     }
     return { interested, notInterested }
   }
@@ -281,7 +281,7 @@ export class StateNetwork extends BaseNetwork {
       node: curRlp,
     })
     await this.db.put(contentKey, dbContent)
-    this.emit('ContentAdded', contentKey, dbContent)
+    this.portal.emit(`${this.networkId}:ContentAdded`, contentKey, dbContent)
   }
 
   async storeStorageTrieNode(contentKey: Uint8Array, content: Uint8Array) {
@@ -291,7 +291,7 @@ export class StateNetwork extends BaseNetwork {
       node: curRlp,
     })
     await this.db.put(contentKey, dbContent)
-    this.emit('ContentAdded', contentKey, dbContent)
+    this.portal.emit(`${this.networkId}:ContentAdded`, contentKey, dbContent)
   }
 
   async receiveStorageTrieNodeOffer(
@@ -362,7 +362,7 @@ export class StateNetwork extends BaseNetwork {
     }
     for (const { contentKey, dbContent } of interested) {
       await this.db.put(contentKey, dbContent)
-      this.emit('ContentAdded', contentKey, dbContent)
+      this.portal.emit(`${this.networkId}:ContentAdded`, contentKey, dbContent)
     }
     return { interested, notInterested }
   }
@@ -394,7 +394,7 @@ export class StateNetwork extends BaseNetwork {
     const codeContent = ContractRetrieval.serialize({ code })
     this.manager.trie.db.local.set(bytesToUnprefixedHex(codeHash), bytesToHex(contentKey))
     await this.db.put(contentKey, codeContent)
-    this.emit('ContentAdded', contentKey, codeContent)
+    this.portal.emit(`${this.networkId}:ContentAdded`, contentKey, codeContent)
     await this.receiveAccountTrieNodeOffer(
       ...extractAccountProof(addressHash, accountProof, blockHash),
     )

packages/portalnetwork/test/integration/beacon.spec.ts

@@ -343,7 +343,7 @@ describe('OFFER/ACCEPT tests', () => {
     )
 
     await new Promise((resolve) => {
-      network2.on('ContentAdded', (contentKey: Uint8Array) => {
+      node2.on(`${network2.networkId}:ContentAdded`, (contentKey: Uint8Array) => {
         const contentType = contentKey[0]
         if (contentType === BeaconNetworkContentType.LightClientOptimisticUpdate)
           // Update the light client stub to report the new "optimistic head"
@@ -522,7 +522,7 @@ describe('OFFER/ACCEPT tests', () => {
     await network1.sendOffer(network2.enr.toENR(), [bootstrapKey])
 
     await new Promise((resolve) => {
-      network2.on('ContentAdded', (key: Uint8Array) => {
+      node2.on(`${network2.networkId}:ContentAdded`, (key: Uint8Array) => {
         assert.deepEqual(key, bootstrapKey, 'successfully gossipped bootstrap')
         resolve(undefined)
       })
@@ -753,7 +753,7 @@ describe('beacon light client sync tests', () => {
     )
 
     await new Promise((resolve) => {
-      network2.portal.on('NodeAdded', (_nodeId) => {
+      node2.on(`${network2.networkId}:NodeAdded`, (_nodeId) => {
         if (network2['bootstrapFinder'].values.length > 0) {
           resolve('undefined)')
         }

packages/portalnetwork/test/integration/ephemeralHeaders.spec.ts

@@ -232,7 +232,7 @@ describe('should offer headers to peers', () => {
     await node1.start()
     await node2.start()
     const history1 = node1.network()['0x500b']
-    const history2 = node2.network()['0x500b']
+    const history2 = node2.network()['0x500b']!
     await history2?.sendPing(node1.discv5.enr.toENR())
     await node1
       .network()
@@ -263,7 +263,7 @@ describe('should offer headers to peers', () => {
     )
     await new Promise((resolve) => {
       let count = 0
-      history2?.on('ContentAdded', async (key, value) => {
+      node2.on(`${history2.networkId}:ContentAdded`, async (key, value) => {
         count++
         if (count === 3) {
           const header = await history2.get(getEphemeralHeaderDbKey(headers[2].hash()))

packages/portalnetwork/test/integration/integration.spec.ts

@@ -94,11 +94,10 @@ describe('gossip test', async () => {
   // Fancy workaround to allow us to "await" an event firing as expected following this - https://github.com/ljharb/tape/pull/503#issuecomment-619358911
   const end = new EventEmitter()
   const addedHeaders: [string, string][] = []
-  network2.on('ContentAdded', async (key: Uint8Array, content: Uint8Array) => {
+  node2.on(`${network2.networkId}:ContentAdded`, async (key: Uint8Array, content: Uint8Array) => {
     network2.logger.extend('ContentAdded')(`Added Content for ${bytesToHex(key)}`)
     addedHeaders.push([bytesToHex(key), bytesToHex(content)])
     if (addedHeaders.length === headersWithProofs.length) {
-      node2.removeAllListeners()
       void node1.stop()
       void node2.stop()
       end.emit('end()')
@@ -218,7 +217,6 @@ describe('FindContent', async () => {
     setHardfork: true,
   })
 
-  node2.removeAllListeners()
   void node1.stop()
   void node2.stop()
   it('should find content', () => {