Enable SSE block_gossip event (#7054)

* feat: add BlockGossipObject for sse event

* feat: enable block_gossip topic

* feat: create async queue and emit event

* chore: copyright year

* fix: check before onAdded exists and remove decodeString

Author: ahshum

beacon_chain/beacon_node.nim

@@ -47,6 +47,7 @@ type
   EventBus* = object
     headQueue*: AsyncEventQueue[HeadChangeInfoObject]
     blocksQueue*: AsyncEventQueue[EventBeaconBlockObject]
+    blockGossipQueue*: AsyncEventQueue[EventBeaconBlockGossipObject]
     phase0AttestQueue*: AsyncEventQueue[phase0.Attestation]
     singleAttestQueue*: AsyncEventQueue[SingleAttestation]
     exitQueue*: AsyncEventQueue[SignedVoluntaryExit]

beacon_chain/consensus_object_pools/block_pools_types.nim

@@ -51,6 +51,8 @@ type
 
   OnBlockCallback* =
     proc(data: ForkedTrustedSignedBeaconBlock) {.gcsafe, raises: [].}
+  OnBlockGossipCallback* =
+    proc(data: ForkedSignedBeaconBlock) {.gcsafe, raises: [].}
   OnHeadCallback* =
     proc(data: HeadChangeInfoObject) {.gcsafe, raises: [].}
   OnReorgCallback* =
@@ -233,6 +235,8 @@ type
 
     onBlockAdded*: OnBlockCallback
       ## On block added callback
+    onBlockGossipAdded*: OnBlockGossipCallback
+      ## On block gossip added callback
     onHeadChanged*: OnHeadCallback
       ## On head changed callback
     onReorgHappened*: OnReorgCallback
@@ -341,6 +345,10 @@ type
     block_root* {.serializedFieldName: "block".}: Eth2Digest
     optimistic* {.serializedFieldName: "execution_optimistic".}: Option[bool]
 
+  EventBeaconBlockGossipObject* = object
+    slot*: Slot
+    block_root* {.serializedFieldName: "block".}: Eth2Digest
+
 template OnBlockAddedCallback*(kind: static ConsensusFork): auto =
   when kind == ConsensusFork.Fulu:
     typedesc[OnFuluBlockAdded]
@@ -401,6 +409,9 @@ template setFinalizationCb*(dag: ChainDAGRef, cb: OnFinalizedCallback) =
 template setBlockCb*(dag: ChainDAGRef, cb: OnBlockCallback) =
   dag.onBlockAdded = cb
 
+template setBlockGossipCb*(dag: ChainDAGRef, cb: OnBlockGossipCallback) =
+  dag.onBlockGossipAdded = cb
+
 template setHeadCb*(dag: ChainDAGRef, cb: OnHeadCallback) =
   dag.onHeadChanged = cb
 
@@ -476,3 +487,11 @@ func init*(t: typedesc[EventBeaconBlockObject],
       block_root: forkyBlck.root,
       optimistic: optimistic
     )
+
+func init*(t: typedesc[EventBeaconBlockGossipObject],
+           v: ForkedSignedBeaconBlock): EventBeaconBlockGossipObject =
+  withBlck(v):
+    EventBeaconBlockGossipObject(
+      slot: forkyBlck.message.slot,
+      block_root: forkyBlck.root
+    )

beacon_chain/gossip_processing/eth2_processor.nim

@@ -1,5 +1,5 @@
 # beacon_chain
-# Copyright (c) 2018-2024 Status Research & Development GmbH
+# Copyright (c) 2018-2025 Status Research & Development GmbH
 # Licensed and distributed under either of
 #   * MIT license (license terms in the root directory or at https://opensource.org/licenses/MIT).
 #   * Apache v2 license (license terms in the root directory or at https://www.apache.org/licenses/LICENSE-2.0).
@@ -236,6 +236,9 @@ proc processSignedBeaconBlock*(
     # propagation of seemingly good blocks
     trace "Block validated"
 
+    if not(isNil(self.dag.onBlockGossipAdded)):
+      self.dag.onBlockGossipAdded(ForkedSignedBeaconBlock.init(signedBlock))
+
     let blobs =
       when typeof(signedBlock).kind >= ConsensusFork.Deneb:
         if self.blobQuarantine[].hasBlobs(signedBlock):

beacon_chain/nimbus_beacon_node.nim

@@ -317,6 +317,9 @@ proc initFullNode(
         none[bool]()
     node.eventBus.blocksQueue.emit(
       EventBeaconBlockObject.init(data, optimistic))
+  proc onBlockGossipAdded(data: ForkedSignedBeaconBlock) =
+    node.eventBus.blockGossipQueue.emit(
+      EventBeaconBlockGossipObject.init(data))
   proc onHeadChanged(data: HeadChangeInfoObject) =
     let eventData =
       if node.currentSlot().epoch() >= dag.cfg.BELLATRIX_FORK_EPOCH:
@@ -594,6 +597,7 @@ proc initFullNode(
 
   dag.setFinalizationCb makeOnFinalizationCb(node.eventBus, node.elManager)
   dag.setBlockCb(onBlockAdded)
+  dag.setBlockGossipCb(onBlockGossipAdded)
   dag.setHeadCb(onHeadChanged)
   dag.setReorgCb(onChainReorg)
 
@@ -739,6 +743,7 @@ proc init*(T: type BeaconNode,
     eventBus = EventBus(
       headQueue: newAsyncEventQueue[HeadChangeInfoObject](),
       blocksQueue: newAsyncEventQueue[EventBeaconBlockObject](),
+      blockGossipQueue: newAsyncEventQueue[EventBeaconBlockGossipObject](),
       phase0AttestQueue: newAsyncEventQueue[phase0.Attestation](),
       singleAttestQueue: newAsyncEventQueue[SingleAttestation](),
       exitQueue: newAsyncEventQueue[SignedVoluntaryExit](),

beacon_chain/rpc/rest_event_api.nim

@@ -125,6 +125,10 @@ proc installEventApiHandlers*(router: var RestRouter, node: BeaconNode) =
           let handler = response.eventHandler(node.eventBus.blocksQueue,
                                               "block")
           res.add(handler)
+        if EventTopic.BlockGossip in eventTopics:
+          let handler = response.eventHandler(node.eventBus.blockGossipQueue,
+                                              "block_gossip")
+          res.add(handler)
         if EventTopic.Attestation in eventTopics:
           let handler = response.eventHandler(node.eventBus.phase0AttestQueue,
                                               "attestation")

beacon_chain/spec/eth2_apis/eth2_rest_serialization.nim

@@ -69,6 +69,7 @@ RestJson.useDefaultSerializationFor(
   EmptyBody,
   Eth1Data,
   EventBeaconBlockObject,
+  EventBeaconBlockGossipObject,
   ExecutionRequests,
   Fork,
   FuluSignedBlockContents,
@@ -3722,6 +3723,8 @@ func decodeString*(t: typedesc[EventTopic],
     ok(EventTopic.Head)
   of "block":
     ok(EventTopic.Block)
+  of "block_gossip":
+    ok(EventTopic.BlockGossip)
   of "attestation":
     ok(EventTopic.Attestation)
   of "single_attestation":
@@ -3755,6 +3758,8 @@ func encodeString*(value: set[EventTopic]): Result[string, cstring] =
     res.add("head,")
   if EventTopic.Block in value:
     res.add("block,")
+  if EventTopic.BlockGossip in value:
+    res.add("block_gossip,")
   if EventTopic.Attestation in value:
     res.add("attestation,")
   if EventTopic.SingleAttestation in value:

beacon_chain/spec/eth2_apis/rest_types.nim

@@ -56,7 +56,7 @@ static:
 type
   # https://github.com/ethereum/beacon-APIs/blob/v2.4.2/apis/eventstream/index.yaml
   EventTopic* {.pure.} = enum
-    Head, Block, Attestation, VoluntaryExit, BLSToExecutionChange,
+    Head, Block, Attestation, BlockGossip, VoluntaryExit, BLSToExecutionChange,
     ProposerSlashing, AttesterSlashing, BlobSidecar, SingleAttestation,
     FinalizedCheckpoint, ChainReorg, ContributionAndProof,
     LightClientFinalityUpdate, LightClientOptimisticUpdate