Add safe block visualization from fast_confirmation event stream

clients/consensus/chainstate.go

@@ -28,6 +28,11 @@ type ChainState struct {
 	finalityMutex sync.RWMutex
 	finality      *v1.Finality
 
+	fastConfirmationMutex sync.RWMutex
+	fastConfirmedSlot     phase0.Slot
+	fastConfirmedRoot     phase0.Root
+	lastFastConfirmation  time.Time
+
 	checkpointDispatcher     utils.Dispatcher[*v1.Finality]
 	wallclockEpochDispatcher utils.Dispatcher[*ethwallclock.Epoch]
 	wallclockSlotDispatcher  utils.Dispatcher[*ethwallclock.Slot]
@@ -229,6 +234,32 @@ func (cs *ChainState) GetGenesis() *v1.Genesis {
 	return cs.genesis
 }
 
+// setFastConfirmedBlock tracks the network-wide safe block as the highest fast confirmed
+// block reported by any client via the fast_confirmation event stream.
+func (cs *ChainState) setFastConfirmedBlock(slot phase0.Slot, root phase0.Root) {
+	cs.fastConfirmationMutex.Lock()
+	defer cs.fastConfirmationMutex.Unlock()
+
+	cs.lastFastConfirmation = time.Now()
+
+	if slot < cs.fastConfirmedSlot {
+		return
+	}
+
+	cs.fastConfirmedSlot = slot
+	cs.fastConfirmedRoot = root
+}
+
+// GetFastConfirmedBlock returns the network-wide safe block (highest fast confirmed block
+// across all clients) and the time of the last fast confirmation event. The returned time
+// is zero if no client ever reported a fast confirmation.
+func (cs *ChainState) GetFastConfirmedBlock() (phase0.Slot, phase0.Root, time.Time) {
+	cs.fastConfirmationMutex.RLock()
+	defer cs.fastConfirmationMutex.RUnlock()
+
+	return cs.fastConfirmedSlot, cs.fastConfirmedRoot, cs.lastFastConfirmation
+}
+
 func (cs *ChainState) GetFinalizedCheckpoint() (phase0.Epoch, phase0.Root) {
 	cs.finalityMutex.RLock()
 	defer cs.finalityMutex.RUnlock()

clients/consensus/client.go

@@ -47,6 +47,9 @@ type Client struct {
 	justifiedEpoch                phase0.Epoch
 	finalizedRoot                 phase0.Root
 	finalizedEpoch                phase0.Epoch
+	fastConfirmedRoot             phase0.Root
+	fastConfirmedSlot             phase0.Slot
+	lastFastConfirmation          time.Time
 	lastFinalityUpdateEpoch       phase0.Epoch
 	lastMetadataUpdateEpoch       phase0.Epoch
 	lastMetadataUpdateTime        time.Time
@@ -57,6 +60,7 @@ type Client struct {
 	executionPayloadDispatcher    utils.Dispatcher[*v1.ExecutionPayloadAvailableEvent]
 	executionPayloadBidDispatcher utils.Dispatcher[*gloas.SignedExecutionPayloadBid]
 	inclusionListDispatcher       utils.Dispatcher[*v1.InclusionListEvent]
+	fastConfirmationDispatcher    utils.Dispatcher[*rpc.FastConfirmationEvent]
 
 	specWarnings []string // warnings from incomplete spec checks
 	specs        map[string]interface{}
@@ -115,6 +119,10 @@ func (client *Client) SubscribeInclusionListEvent(capacity int, blocking bool) *
 	return client.inclusionListDispatcher.Subscribe(capacity, blocking)
 }
 
+func (client *Client) SubscribeFastConfirmationEvent(capacity int, blocking bool) *utils.Subscription[*rpc.FastConfirmationEvent] {
+	return client.fastConfirmationDispatcher.Subscribe(capacity, blocking)
+}
+
 func (client *Client) GetPool() *Pool {
 	return client.pool
 }
@@ -173,6 +181,16 @@ func (client *Client) GetFinalityCheckpoint() (finalitedEpoch phase0.Epoch, fina
 	return client.finalizedEpoch, client.finalizedRoot, client.justifiedEpoch, client.justifiedRoot
 }
 
+// GetLastFastConfirmation returns the most recent fast confirmed (safe) block reported
+// by the node via the fast_confirmation event stream. The returned time is the time the
+// last fast confirmation event was received (zero if the node never sent one).
+func (client *Client) GetLastFastConfirmation() (phase0.Slot, phase0.Root, time.Time) {
+	client.headMutex.RLock()
+	defer client.headMutex.RUnlock()
+
+	return client.fastConfirmedSlot, client.fastConfirmedRoot, client.lastFastConfirmation
+}
+
 func (client *Client) GetStatus() ClientStatus {
 	switch {
 	case client.isSyncing:

clients/consensus/clientlogic.go

@@ -130,6 +130,12 @@ func (client *Client) buildEventStreamMask() uint16 {
 		events |= rpc.StreamInclusionListEvent
 	}
 
+	// fast confirmation is an optional node feature. nodes reject the whole
+	// subscription with a HTTP 400 for unknown topics, so the topic is requested
+	// via a separate SSE stream that silently gives up on rejection instead of
+	// being mixed into the main block/head/finalized stream
+	events |= rpc.StreamFastConfirmationEvent
+
 	return events
 }
 
@@ -191,6 +197,9 @@ func (client *Client) runClientLogic() error {
 
 			case rpc.StreamInclusionListEvent:
 				client.inclusionListDispatcher.Fire(evt.Data.(*v1.InclusionListEvent))
+
+			case rpc.StreamFastConfirmationEvent:
+				client.processFastConfirmationEvent(evt.Data.(*rpc.FastConfirmationEvent))
 			}
 
 			// fire through stream dispatcher first to preserve SSE ordering
@@ -363,6 +372,17 @@ func (client *Client) processHeadEvent(evt *v1.HeadEvent) error {
 	return nil
 }
 
+func (client *Client) processFastConfirmationEvent(evt *rpc.FastConfirmationEvent) {
+	client.headMutex.Lock()
+	client.fastConfirmedSlot = evt.Slot
+	client.fastConfirmedRoot = evt.Block
+	client.lastFastConfirmation = time.Now()
+	client.headMutex.Unlock()
+
+	client.pool.chainState.setFastConfirmedBlock(evt.Slot, evt.Block)
+	client.fastConfirmationDispatcher.Fire(evt)
+}
+
 func (client *Client) processFinalizedEvent(evt *v1.FinalizedCheckpointEvent) error {
 	go func() {
 		retry := 0

clients/consensus/rpc/beaconstream.go

@@ -2,16 +2,20 @@ package rpc
 
 import (
 	"context"
+	"encoding/hex"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"net/http"
 	"net/url"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
 
 	v1 "github.com/ethpandaops/go-eth2-client/api/v1"
 	"github.com/ethpandaops/go-eth2-client/spec/gloas"
+	"github.com/ethpandaops/go-eth2-client/spec/phase0"
 	"github.com/sirupsen/logrus"
 
 	"github.com/ethpandaops/dora/clients/consensus/rpc/eventstream"
@@ -24,13 +28,65 @@ const (
 	StreamExecutionPayloadEvent    uint16 = 0x08
 	StreamExecutionPayloadBidEvent uint16 = 0x10
 	StreamInclusionListEvent       uint16 = 0x20
+	StreamFastConfirmationEvent    uint16 = 0x40
 )
 
 type BeaconStreamEvent struct {
 	Event uint16
 	Data  interface{}
 }
 
+// FastConfirmationEvent is the payload of the fast_confirmation event stream topic.
+// Slot/Block identify the most recent fast confirmed (safe) block, CurrentSlot is the
+// wall-clock slot at which the fast confirmation rule was executed (optional, added
+// by ethereum/beacon-APIs#616).
+type FastConfirmationEvent struct {
+	Slot        phase0.Slot
+	Block       phase0.Root
+	CurrentSlot phase0.Slot
+}
+
+func (e *FastConfirmationEvent) UnmarshalJSON(input []byte) error {
+	var data struct {
+		Slot        string `json:"slot"`
+		Block       string `json:"block"`
+		CurrentSlot string `json:"current_slot"`
+	}
+
+	if err := json.Unmarshal(input, &data); err != nil {
+		return err
+	}
+
+	slot, err := strconv.ParseUint(data.Slot, 10, 64)
+	if err != nil {
+		return fmt.Errorf("invalid value for slot: %w", err)
+	}
+
+	e.Slot = phase0.Slot(slot)
+
+	block, err := hex.DecodeString(strings.TrimPrefix(data.Block, "0x"))
+	if err != nil {
+		return fmt.Errorf("invalid value for block: %w", err)
+	}
+
+	if len(block) != len(e.Block) {
+		return fmt.Errorf("incorrect length %d for block", len(block))
+	}
+
+	copy(e.Block[:], block)
+
+	if data.CurrentSlot != "" {
+		currentSlot, err := strconv.ParseUint(data.CurrentSlot, 10, 64)
+		if err != nil {
+			return fmt.Errorf("invalid value for current_slot: %w", err)
+		}
+
+		e.CurrentSlot = phase0.Slot(currentSlot)
+	}
+
+	return nil
+}
+
 type BeaconStreamStatus struct {
 	Ready bool
 	Error error
@@ -78,7 +134,7 @@ func (bs *BeaconStream) startStream() {
 	}()
 
 	basicEvents := bs.events & (StreamBlockEvent | StreamHeadEvent | StreamFinalizedEvent)
-	basicStream := bs.subscribeStream(bs.client.endpoint, basicEvents)
+	basicStream := bs.subscribeStream(bs.client.endpoint, basicEvents, false)
 	if basicStream == nil {
 		return
 	}
@@ -125,6 +181,11 @@ func (bs *BeaconStream) ensureAncillaryStreams(events uint16) {
 
 	hezeEvents := events & StreamInclusionListEvent
 	bs.startAncillaryStream(hezeEvents)
+
+	// fast confirmation is an optional node feature (not fork gated), so it gets its
+	// own stream that silently gives up if the node rejects the topic
+	fastConfirmationEvents := events & StreamFastConfirmationEvent
+	bs.startAncillaryStream(fastConfirmationEvents)
 }
 
 func (bs *BeaconStream) startAncillaryStream(events uint16) {
@@ -144,7 +205,10 @@ func (bs *BeaconStream) startAncillaryStream(events uint16) {
 }
 
 func (bs *BeaconStream) runAncillaryStream(events uint16) {
-	stream := bs.subscribeStream(bs.client.endpoint, events)
+	// streams carrying only optional topics give up when the node rejects the topic
+	optional := events&StreamFastConfirmationEvent == events
+
+	stream := bs.subscribeStream(bs.client.endpoint, events, optional)
 	if stream == nil {
 		return
 	}
@@ -162,15 +226,33 @@ func (bs *BeaconStream) runAncillaryStream(events uint16) {
 				bs.processExecutionPayloadBidEvent(evt)
 			case "inclusion_list":
 				bs.processInclusionListEvent(evt)
+			case "fast_confirmation":
+				bs.processFastConfirmationEvent(evt)
 			}
 		case <-stream.Ready:
-		case <-stream.Errors:
+		case err := <-stream.Errors:
+			if optional && isUnsupportedTopicError(err) {
+				bs.logger.Debugf("optional beacon event stream not supported by node (events: 0x%x): %v", events, err)
+				return
+			}
+
 			time.Sleep(10 * time.Millisecond)
 			stream.RetryNow()
 		}
 	}
 }
 
+// isUnsupportedTopicError checks if the given stream error indicates that the node
+// rejected the event subscription (4xx response, eg. unsupported topics).
+func isUnsupportedTopicError(err error) bool {
+	var subErr eventstream.SubscriptionError
+	if errors.As(err, &subErr) {
+		return subErr.Code >= 400 && subErr.Code < 500
+	}
+
+	return false
+}
+
 // handleStreamError handles stream errors and forwards them to the ReadyChan.
 func (bs *BeaconStream) handleStreamError(stream *eventstream.Stream, err error) {
 	if strings.Contains(err.Error(), "INTERNAL_ERROR; received from peer") {
@@ -190,7 +272,7 @@ func (bs *BeaconStream) handleStreamError(stream *eventstream.Stream, err error)
 	}
 }
 
-func (bs *BeaconStream) subscribeStream(endpoint string, events uint16) *eventstream.Stream {
+func (bs *BeaconStream) subscribeStream(endpoint string, events uint16, optional bool) *eventstream.Stream {
 	var topics strings.Builder
 
 	topicsCount := 0
@@ -255,6 +337,16 @@ func (bs *BeaconStream) subscribeStream(endpoint string, events uint16) *eventst
 		topicsCount++
 	}
 
+	if events&StreamFastConfirmationEvent > 0 {
+		if topicsCount > 0 {
+			fmt.Fprintf(&topics, ",")
+		}
+
+		fmt.Fprintf(&topics, "fast_confirmation")
+
+		topicsCount++
+	}
+
 	if topicsCount == 0 {
 		return nil
 	}
@@ -274,6 +366,11 @@ func (bs *BeaconStream) subscribeStream(endpoint string, events uint16) *eventst
 		}
 
 		if err != nil {
+			if optional && isUnsupportedTopicError(err) {
+				bs.logger.Debugf("optional beacon event stream %v not supported by node: %v", getRedactedURL(streamURL), err)
+				return nil
+			}
+
 			bs.logger.Warnf("Error while subscribing beacon event stream %v: %v", getRedactedURL(streamURL), err)
 			select {
 			case <-bs.ctx.Done():
@@ -370,6 +467,21 @@ func (bs *BeaconStream) processExecutionPayloadBidEvent(evt eventstream.StreamEv
 	}
 }
 
+func (bs *BeaconStream) processFastConfirmationEvent(evt eventstream.StreamEvent) {
+	var parsed FastConfirmationEvent
+
+	err := json.Unmarshal([]byte(evt.Data()), &parsed)
+	if err != nil {
+		bs.logger.Warnf("beacon block stream failed to decode fast_confirmation event: %v", err)
+		return
+	}
+
+	bs.EventChan <- &BeaconStreamEvent{
+		Event: StreamFastConfirmationEvent,
+		Data:  &parsed,
+	}
+}
+
 func (bs *BeaconStream) processInclusionListEvent(evt eventstream.StreamEvent) {
 	var parsed v1.InclusionListEvent
 

cmd/dora-explorer/main.go

@@ -353,6 +353,7 @@ func startApi(router *mux.Router) {
 		{"/v1/network/forks", api.APINetworkForksV1, []string{"GET", "OPTIONS"}, 1},
 		{"/v1/network/splits", api.APINetworkSplitsV1, []string{"GET", "OPTIONS"}, 1},
 		{"/v1/network/client_head_forks", api.APINetworkClientHeadForksV1, []string{"GET", "OPTIONS"}, 1},
+		{"/v1/network/fast_confirmation", api.APINetworkFastConfirmationV1, []string{"GET", "OPTIONS"}, 1},
 
 		// Client APIs
 		{"/v1/clients/execution", api.APIExecutionClients, []string{"GET", "OPTIONS"}, 1},