emit proposer_slashing/attester_slashing SSE on beacon-API (#5678)

Add support for slashings on the beacon-API event stream for compat with
beacon-API specs.

- ethereum/beacon-APIs#376

beacon_chain/beacon_node.nim

@@ -45,6 +45,8 @@ type
     attestQueue*: AsyncEventQueue[Attestation]
     exitQueue*: AsyncEventQueue[SignedVoluntaryExit]
     blsToExecQueue*: AsyncEventQueue[SignedBLSToExecutionChange]
+    propSlashQueue*: AsyncEventQueue[ProposerSlashing]
+    attSlashQueue*: AsyncEventQueue[AttesterSlashing]
     finalQueue*: AsyncEventQueue[FinalizationInfoObject]
     reorgQueue*: AsyncEventQueue[ReorgInfoObject]
     contribQueue*: AsyncEventQueue[SignedContributionAndProof]

beacon_chain/consensus_object_pools/exit_pool.nim

@@ -32,6 +32,10 @@ type
     proc(data: SignedVoluntaryExit) {.gcsafe, raises: [].}
   OnBLSToExecutionChangeCallback =
     proc(data: SignedBLSToExecutionChange) {.gcsafe, raises: [].}
+  OnProposerSlashingCallback =
+    proc(data: ProposerSlashing) {.gcsafe, raises: [].}
+  OnAttesterSlashingCallback =
+    proc(data: AttesterSlashing) {.gcsafe, raises: [].}
 
   ValidatorChangePool* = object
     ## The validator change pool tracks attester slashings, proposer slashings,
@@ -69,11 +73,15 @@ type
     attestationPool: ref AttestationPool
     onVoluntaryExitReceived*: OnVoluntaryExitCallback
     onBLSToExecutionChangeReceived*: OnBLSToExecutionChangeCallback
+    onProposerSlashingReceived*: OnProposerSlashingCallback
+    onAttesterSlashingReceived*: OnAttesterSlashingCallback
 
 func init*(T: type ValidatorChangePool, dag: ChainDAGRef,
            attestationPool: ref AttestationPool = nil,
            onVoluntaryExit: OnVoluntaryExitCallback = nil,
-           onBLSToExecutionChange: OnBLSToExecutionChangeCallback = nil): T =
+           onBLSToExecutionChange: OnBLSToExecutionChangeCallback = nil,
+           onProposerSlashing: OnProposerSlashingCallback = nil,
+           onAttesterSlashing: OnAttesterSlashingCallback = nil): T =
   ## Initialize an ValidatorChangePool from the dag `headState`
   T(
     # Allow filtering some validator change messages during block production
@@ -96,7 +104,9 @@ func init*(T: type ValidatorChangePool, dag: ChainDAGRef,
     dag: dag,
     attestationPool: attestationPool,
     onVoluntaryExitReceived: onVoluntaryExit,
-    onBLSToExecutionChangeReceived: onBLSToExecutionChange)
+    onBLSToExecutionChangeReceived: onBLSToExecutionChange,
+    onProposerSlashingReceived: onProposerSlashing,
+    onAttesterSlashingReceived: onAttesterSlashing)
 
 func addValidatorChangeMessage(
     subpool: var auto, seenpool: var auto, validatorChangeMessage: auto,

beacon_chain/gossip_processing/gossip_validation.nim

@@ -1075,6 +1075,10 @@ proc validateAttesterSlashing*(
   if attester_slashing_validity.isErr:
     return pool.checkedReject(attester_slashing_validity.error)
 
+  # Send notification about new attester slashing via callback
+  if not(isNil(pool.onAttesterSlashingReceived)):
+    pool.onAttesterSlashingReceived(attester_slashing)
+
   ok()
 
 # https://github.com/ethereum/consensus-specs/blob/v1.4.0-beta.4/specs/phase0/p2p-interface.md#proposer_slashing
@@ -1099,6 +1103,10 @@ proc validateProposerSlashing*(
   if proposer_slashing_validity.isErr:
     return pool.checkedReject(proposer_slashing_validity.error)
 
+  # Send notification about new proposer slashing via callback
+  if not(isNil(pool.onProposerSlashingReceived)):
+    pool.onProposerSlashingReceived(proposer_slashing)
+
   ok()
 
 # https://github.com/ethereum/consensus-specs/blob/v1.4.0-beta.5/specs/phase0/p2p-interface.md#voluntary_exit

beacon_chain/nimbus_beacon_node.nim

@@ -289,6 +289,10 @@ proc initFullNode(
     node.eventBus.exitQueue.emit(data)
   proc onBLSToExecutionChangeAdded(data: SignedBLSToExecutionChange) =
     node.eventBus.blsToExecQueue.emit(data)
+  proc onProposerSlashingAdded(data: ProposerSlashing) =
+    node.eventBus.propSlashQueue.emit(data)
+  proc onAttesterSlashingAdded(data: AttesterSlashing) =
+    node.eventBus.attSlashQueue.emit(data)
   proc onBlockAdded(data: ForkedTrustedSignedBeaconBlock) =
     let optimistic =
       if node.currentSlot().epoch() >= dag.cfg.BELLATRIX_FORK_EPOCH:
@@ -367,7 +371,8 @@ proc initFullNode(
     lightClientPool = newClone(
       LightClientPool())
     validatorChangePool = newClone(ValidatorChangePool.init(
-      dag, attestationPool, onVoluntaryExitAdded, onBLSToExecutionChangeAdded))
+      dag, attestationPool, onVoluntaryExitAdded, onBLSToExecutionChangeAdded,
+      onProposerSlashingAdded, onAttesterSlashingAdded))
     blobQuarantine = newClone(BlobQuarantine())
     consensusManager = ConsensusManager.new(
       dag, attestationPool, quarantine, node.elManager,
@@ -546,6 +551,8 @@ proc init*(T: type BeaconNode,
       attestQueue: newAsyncEventQueue[Attestation](),
       exitQueue: newAsyncEventQueue[SignedVoluntaryExit](),
       blsToExecQueue: newAsyncEventQueue[SignedBLSToExecutionChange](),
+      propSlashQueue: newAsyncEventQueue[ProposerSlashing](),
+      attSlashQueue: newAsyncEventQueue[AttesterSlashing](),
       finalQueue: newAsyncEventQueue[FinalizationInfoObject](),
       reorgQueue: newAsyncEventQueue[ReorgInfoObject](),
       contribQueue: newAsyncEventQueue[SignedContributionAndProof](),

beacon_chain/rpc/rest_event_api.nim

@@ -137,6 +137,14 @@ proc installEventApiHandlers*(router: var RestRouter, node: BeaconNode) =
           let handler = response.eventHandler(node.eventBus.blsToExecQueue,
                                               "bls_to_execution_change")
           res.add(handler)
+        if EventTopic.ProposerSlashing in eventTopics:
+          let handler = response.eventHandler(node.eventBus.propSlashQueue,
+                                              "proposer_slashing")
+          res.add(handler)
+        if EventTopic.AttesterSlashing in eventTopics:
+          let handler = response.eventHandler(node.eventBus.attSlashQueue,
+                                              "attester_slashing")
+          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

@@ -4141,6 +4141,10 @@ proc decodeString*(t: typedesc[EventTopic],
     ok(EventTopic.VoluntaryExit)
   of "bls_to_execution_change":
     ok(EventTopic.BLSToExecutionChange)
+  of "proposer_slashing":
+    ok(EventTopic.ProposerSlashing)
+  of "attester_slashing":
+    ok(EventTopic.AttesterSlashing)
   of "finalized_checkpoint":
     ok(EventTopic.FinalizedCheckpoint)
   of "chain_reorg":
@@ -4166,6 +4170,10 @@ proc encodeString*(value: set[EventTopic]): Result[string, cstring] =
     res.add("voluntary_exit,")
   if EventTopic.BLSToExecutionChange in value:
     res.add("bls_to_execution_change,")
+  if EventTopic.ProposerSlashing in value:
+    res.add("proposer_slashing,")
+  if EventTopic.AttesterSlashing in value:
+    res.add("attester_slashing,")
   if EventTopic.FinalizedCheckpoint in value:
     res.add("finalized_checkpoint,")
   if EventTopic.ChainReorg in value:

beacon_chain/spec/eth2_apis/rest_types.nim

@@ -55,8 +55,8 @@ type
   # https://github.com/ethereum/beacon-APIs/blob/v2.4.2/apis/eventstream/index.yaml
   EventTopic* {.pure.} = enum
     Head, Block, Attestation, VoluntaryExit, BLSToExecutionChange,
-    FinalizedCheckpoint, ChainReorg, ContributionAndProof,
-    LightClientFinalityUpdate, LightClientOptimisticUpdate
+    ProposerSlashing, AttesterSlashing, FinalizedCheckpoint, ChainReorg,
+    ContributionAndProof, LightClientFinalityUpdate, LightClientOptimisticUpdate
 
   EventTopics* = set[EventTopic]