emit bls_to_execution_change SSE on beacon-API (#5677)

With Capella, `bls_to_execution_change` SSE should be emitted on the
event stream whenever a new `SignedBLSToExecutionChange` is received.
Add this missing functionality for compatibility with beacon-API specs.

- ethereum/beacon-APIs#248

beacon_chain/beacon_node.nim

@@ -40,17 +40,18 @@ export
 
 type
   EventBus* = object
-    blocksQueue*: AsyncEventQueue[EventBeaconBlockObject]
     headQueue*: AsyncEventQueue[HeadChangeInfoObject]
+    blocksQueue*: AsyncEventQueue[EventBeaconBlockObject]
+    attestQueue*: AsyncEventQueue[Attestation]
+    exitQueue*: AsyncEventQueue[SignedVoluntaryExit]
+    blsToExecQueue*: AsyncEventQueue[SignedBLSToExecutionChange]
+    finalQueue*: AsyncEventQueue[FinalizationInfoObject]
     reorgQueue*: AsyncEventQueue[ReorgInfoObject]
+    contribQueue*: AsyncEventQueue[SignedContributionAndProof]
     finUpdateQueue*: AsyncEventQueue[
       RestVersioned[ForkedLightClientFinalityUpdate]]
     optUpdateQueue*: AsyncEventQueue[
       RestVersioned[ForkedLightClientOptimisticUpdate]]
-    attestQueue*: AsyncEventQueue[Attestation]
-    contribQueue*: AsyncEventQueue[SignedContributionAndProof]
-    exitQueue*: AsyncEventQueue[SignedVoluntaryExit]
-    finalQueue*: AsyncEventQueue[FinalizationInfoObject]
 
   BeaconNode* = ref object
     nickname*: string

beacon_chain/consensus_object_pools/exit_pool.nim

@@ -30,6 +30,8 @@ const
 type
   OnVoluntaryExitCallback =
     proc(data: SignedVoluntaryExit) {.gcsafe, raises: [].}
+  OnBLSToExecutionChangeCallback =
+    proc(data: SignedBLSToExecutionChange) {.gcsafe, raises: [].}
 
   ValidatorChangePool* = object
     ## The validator change pool tracks attester slashings, proposer slashings,
@@ -66,10 +68,12 @@ type
     dag*: ChainDAGRef
     attestationPool: ref AttestationPool
     onVoluntaryExitReceived*: OnVoluntaryExitCallback
+    onBLSToExecutionChangeReceived*: OnBLSToExecutionChangeCallback
 
 func init*(T: type ValidatorChangePool, dag: ChainDAGRef,
            attestationPool: ref AttestationPool = nil,
-           onVoluntaryExit: OnVoluntaryExitCallback = nil): T =
+           onVoluntaryExit: OnVoluntaryExitCallback = nil,
+           onBLSToExecutionChange: OnBLSToExecutionChangeCallback = nil): T =
   ## Initialize an ValidatorChangePool from the dag `headState`
   T(
     # Allow filtering some validator change messages during block production
@@ -91,8 +95,8 @@ func init*(T: type ValidatorChangePool, dag: ChainDAGRef,
       initDeque[SignedBLSToExecutionChange](initialSize = 1024),
     dag: dag,
     attestationPool: attestationPool,
-    onVoluntaryExitReceived: onVoluntaryExit
-   )
+    onVoluntaryExitReceived: onVoluntaryExit,
+    onBLSToExecutionChangeReceived: onBLSToExecutionChange)
 
 func addValidatorChangeMessage(
     subpool: var auto, seenpool: var auto, validatorChangeMessage: auto,

beacon_chain/gossip_processing/gossip_validation.nim

@@ -1050,6 +1050,10 @@ proc validateBlsToExecutionChange*(
       of BatchResult.Valid:
         discard  # keep going only in this case
 
+  # Send notification about new BLS to execution change via callback
+  if not(isNil(pool.onBLSToExecutionChangeReceived)):
+    pool.onBLSToExecutionChangeReceived(signed_address_change)
+
   return ok()
 
 # https://github.com/ethereum/consensus-specs/blob/v1.3.0/specs/phase0/p2p-interface.md#attester_slashing

beacon_chain/nimbus_beacon_node.nim

@@ -287,6 +287,8 @@ proc initFullNode(
     node.eventBus.contribQueue.emit(data)
   proc onVoluntaryExitAdded(data: SignedVoluntaryExit) =
     node.eventBus.exitQueue.emit(data)
+  proc onBLSToExecutionChangeAdded(data: SignedBLSToExecutionChange) =
+    node.eventBus.blsToExecQueue.emit(data)
   proc onBlockAdded(data: ForkedTrustedSignedBeaconBlock) =
     let optimistic =
       if node.currentSlot().epoch() >= dag.cfg.BELLATRIX_FORK_EPOCH:
@@ -364,8 +366,8 @@ proc initFullNode(
       SyncCommitteeMsgPool.init(rng, dag.cfg, onSyncContribution))
     lightClientPool = newClone(
       LightClientPool())
-    validatorChangePool = newClone(
-      ValidatorChangePool.init(dag, attestationPool, onVoluntaryExitAdded))
+    validatorChangePool = newClone(ValidatorChangePool.init(
+      dag, attestationPool, onVoluntaryExitAdded, onBLSToExecutionChangeAdded))
     blobQuarantine = newClone(BlobQuarantine())
     consensusManager = ConsensusManager.new(
       dag, attestationPool, quarantine, node.elManager,
@@ -539,18 +541,18 @@ proc init*(T: type BeaconNode,
 
   let
     eventBus = EventBus(
-      blocksQueue: newAsyncEventQueue[EventBeaconBlockObject](),
       headQueue: newAsyncEventQueue[HeadChangeInfoObject](),
+      blocksQueue: newAsyncEventQueue[EventBeaconBlockObject](),
+      attestQueue: newAsyncEventQueue[Attestation](),
+      exitQueue: newAsyncEventQueue[SignedVoluntaryExit](),
+      blsToExecQueue: newAsyncEventQueue[SignedBLSToExecutionChange](),
+      finalQueue: newAsyncEventQueue[FinalizationInfoObject](),
       reorgQueue: newAsyncEventQueue[ReorgInfoObject](),
+      contribQueue: newAsyncEventQueue[SignedContributionAndProof](),
       finUpdateQueue: newAsyncEventQueue[
         RestVersioned[ForkedLightClientFinalityUpdate]](),
       optUpdateQueue: newAsyncEventQueue[
-        RestVersioned[ForkedLightClientOptimisticUpdate]](),
-      attestQueue: newAsyncEventQueue[Attestation](),
-      contribQueue: newAsyncEventQueue[SignedContributionAndProof](),
-      exitQueue: newAsyncEventQueue[SignedVoluntaryExit](),
-      finalQueue: newAsyncEventQueue[FinalizationInfoObject]()
-    )
+        RestVersioned[ForkedLightClientOptimisticUpdate]]())
     db = BeaconChainDB.new(config.databaseDir, cfg, inMemory = false)
 
   if config.externalBeaconApiUrl.isSome and ChainDAGRef.isInitialized(db).isErr:

beacon_chain/rpc/rest_event_api.nim

@@ -133,6 +133,10 @@ proc installEventApiHandlers*(router: var RestRouter, node: BeaconNode) =
           let handler = response.eventHandler(node.eventBus.exitQueue,
                                               "voluntary_exit")
           res.add(handler)
+        if EventTopic.BLSToExecutionChange in eventTopics:
+          let handler = response.eventHandler(node.eventBus.blsToExecQueue,
+                                              "bls_to_execution_change")
+          res.add(handler)
         if EventTopic.FinalizedCheckpoint in eventTopics:
           let handler = response.eventHandler(node.eventBus.finalQueue,
                                               "finalized_checkpoint")

beacon_chain/spec/eth2_apis/eth2_rest_serialization.nim

@@ -4139,6 +4139,8 @@ proc decodeString*(t: typedesc[EventTopic],
     ok(EventTopic.Attestation)
   of "voluntary_exit":
     ok(EventTopic.VoluntaryExit)
+  of "bls_to_execution_change":
+    ok(EventTopic.BLSToExecutionChange)
   of "finalized_checkpoint":
     ok(EventTopic.FinalizedCheckpoint)
   of "chain_reorg":
@@ -4162,6 +4164,8 @@ proc encodeString*(value: set[EventTopic]): Result[string, cstring] =
     res.add("attestation,")
   if EventTopic.VoluntaryExit in value:
     res.add("voluntary_exit,")
+  if EventTopic.BLSToExecutionChange in value:
+    res.add("bls_to_execution_change,")
   if EventTopic.FinalizedCheckpoint in value:
     res.add("finalized_checkpoint,")
   if EventTopic.ChainReorg in value:

beacon_chain/spec/eth2_apis/rest_types.nim

@@ -52,9 +52,11 @@ static:
   doAssert(ClientMaximumValidatorIds <= ServerMaximumValidatorIds)
 
 type
+  # https://github.com/ethereum/beacon-APIs/blob/v2.4.2/apis/eventstream/index.yaml
   EventTopic* {.pure.} = enum
-    Head, Block, Attestation, VoluntaryExit, FinalizedCheckpoint, ChainReorg,
-    ContributionAndProof, LightClientFinalityUpdate, LightClientOptimisticUpdate
+    Head, Block, Attestation, VoluntaryExit, BLSToExecutionChange,
+    FinalizedCheckpoint, ChainReorg, ContributionAndProof,
+    LightClientFinalityUpdate, LightClientOptimisticUpdate
 
   EventTopics* = set[EventTopic]