Add fast_confirmation SSE event (beacon-APIs#598)

beacon_node/beacon_chain/src/canonical_head.rs

@@ -45,7 +45,9 @@ use crate::{
     metrics,
     validator_monitor::get_slot_delay_ms,
 };
-use eth2::types::{EventKind, SseChainReorg, SseFinalizedCheckpoint, SseLateHead};
+use eth2::types::{
+    EventKind, SseChainReorg, SseFastConfirmation, SseFinalizedCheckpoint, SseLateHead,
+};
 use fork_choice::{
     ExecutionStatus, ForkChoiceStore, ForkChoiceView, ForkchoiceUpdateParameters, PayloadStatus,
     ProtoBlock, ResetPayloadStatuses,
@@ -741,15 +743,16 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 let label: &'static str = (&e).into();
                 metrics::inc_counter_vec(&metrics::FCR_ERRORS, &[label]);
             } else {
-                if fcr.confirmed_root != old_confirmed {
+                let confirmed_root_changed = fcr.confirmed_root != old_confirmed;
+                if confirmed_root_changed {
                     metrics::inc_counter(&metrics::FCR_CONFIRMED_ROOT_CHANGES);
                 }
-                if let Some(confirmed_slot) = proto_array
+                let confirmed_slot_opt = proto_array
                     .indices
                     .get(&fcr.confirmed_root)
                     .and_then(|&idx| proto_array.nodes.get(idx))
-                    .map(|n| n.slot())
-                {
+                    .map(|n| n.slot());
+                if let Some(confirmed_slot) = confirmed_slot_opt {
                     metrics::set_gauge(
                         &metrics::FCR_CONFIRMED_ROOT_SLOT,
                         confirmed_slot.as_u64() as i64,
@@ -759,6 +762,37 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                         .saturating_sub(confirmed_slot.as_u64());
                     metrics::set_gauge(&metrics::FCR_CONFIRMATION_DELAY_SLOTS, delay as i64);
                 }
+                if confirmed_root_changed
+                    && confirmed_slot_opt.is_some()
+                    && let Some(event_handler) = self.event_handler.as_ref()
+                    && event_handler.has_fast_confirmation_subscribers()
+                {
+                    // Emit one event per newly-confirmed block, oldest-to-newest, so
+                    // consumers see every slot FCR advanced over (not just the tip). The
+                    // depth cap bounds the rare case where old_confirmed is not in
+                    // proto_array (e.g., pruned across a restart).
+                    const MAX_CHAIN_DEPTH: usize = 4096;
+                    let old_confirmed_slot = proto_array
+                        .indices
+                        .get(&old_confirmed)
+                        .and_then(|&idx| proto_array.nodes.get(idx))
+                        .map(|n| n.slot());
+                    let mut newly_confirmed: Vec<(Hash256, Slot)> = proto_array
+                        .iter_block_roots(&fcr.confirmed_root)
+                        .take(MAX_CHAIN_DEPTH)
+                        .take_while(|(root, slot)| match old_confirmed_slot {
+                            Some(old_slot) => *slot > old_slot,
+                            None => *root != old_confirmed,
+                        })
+                        .collect();
+                    newly_confirmed.reverse();
+                    for (block, slot) in newly_confirmed {
+                        event_handler.register(EventKind::FastConfirmation(SseFastConfirmation {
+                            block,
+                            slot,
+                        }));
+                    }
+                }
                 let balance_epoch = fcr.current_balance_source.checkpoint.epoch;
                 let current_epoch = current_slot.epoch(T::EthSpec::slots_per_epoch());
                 let age = current_epoch

beacon_node/beacon_chain/src/events.rs

@@ -30,6 +30,7 @@ pub struct ServerSentEventHandler<E: EthSpec> {
     execution_payload_available_tx: Sender<EventKind<E>>,
     execution_payload_bid_tx: Sender<EventKind<E>>,
     payload_attestation_message_tx: Sender<EventKind<E>>,
+    fast_confirmation_tx: Sender<EventKind<E>>,
 }
 
 impl<E: EthSpec> ServerSentEventHandler<E> {
@@ -61,6 +62,7 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
         let (execution_payload_available_tx, _) = broadcast::channel(capacity);
         let (execution_payload_bid_tx, _) = broadcast::channel(capacity);
         let (payload_attestation_message_tx, _) = broadcast::channel(capacity);
+        let (fast_confirmation_tx, _) = broadcast::channel(capacity);
 
         Self {
             attestation_tx,
@@ -86,6 +88,7 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
             execution_payload_available_tx,
             execution_payload_bid_tx,
             payload_attestation_message_tx,
+            fast_confirmation_tx,
         }
     }
 
@@ -190,6 +193,10 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
                 .payload_attestation_message_tx
                 .send(kind)
                 .map(|count| log_count("payload attestation message", count)),
+            EventKind::FastConfirmation(_) => self
+                .fast_confirmation_tx
+                .send(kind)
+                .map(|count| log_count("fast confirmation", count)),
         };
         if let Err(SendError(event)) = result {
             trace!(?event, "No receivers registered to listen for event");
@@ -288,6 +295,10 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
         self.payload_attestation_message_tx.subscribe()
     }
 
+    pub fn subscribe_fast_confirmation(&self) -> Receiver<EventKind<E>> {
+        self.fast_confirmation_tx.subscribe()
+    }
+
     pub fn has_attestation_subscribers(&self) -> bool {
         self.attestation_tx.receiver_count() > 0
     }
@@ -371,4 +382,8 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
     pub fn has_payload_attestation_message_subscribers(&self) -> bool {
         self.payload_attestation_message_tx.receiver_count() > 0
     }
+
+    pub fn has_fast_confirmation_subscribers(&self) -> bool {
+        self.fast_confirmation_tx.receiver_count() > 0
+    }
 }

beacon_node/http_api/src/lib.rs

@@ -3244,6 +3244,9 @@ pub fn serve<T: BeaconChainTypes>(
                                 api_types::EventTopic::PayloadAttestationMessage => {
                                     event_handler.subscribe_payload_attestation_message()
                                 }
+                                api_types::EventTopic::FastConfirmation => {
+                                    event_handler.subscribe_fast_confirmation()
+                                }
                             };
 
                             receivers.push(

common/eth2/src/types.rs

@@ -1073,6 +1073,12 @@ pub struct BlockGossip {
     pub slot: Slot,
     pub block: Hash256,
 }
+
+#[derive(PartialEq, Debug, Serialize, Deserialize, Clone)]
+pub struct SseFastConfirmation {
+    pub block: Hash256,
+    pub slot: Slot,
+}
 #[derive(PartialEq, Debug, Serialize, Deserialize, Clone)]
 pub struct SseExecutionPayload {
     pub slot: Slot,
@@ -1245,6 +1251,7 @@ pub enum EventKind<E: EthSpec> {
     ExecutionPayloadAvailable(SseExecutionPayloadAvailable),
     ExecutionPayloadBid(Box<VersionedSseExecutionPayloadBid<E>>),
     PayloadAttestationMessage(Box<VersionedSsePayloadAttestationMessage>),
+    FastConfirmation(SseFastConfirmation),
 }
 
 impl<E: EthSpec> EventKind<E> {
@@ -1273,6 +1280,7 @@ impl<E: EthSpec> EventKind<E> {
             EventKind::ExecutionPayloadAvailable(_) => "execution_payload_available",
             EventKind::ExecutionPayloadBid(_) => "execution_payload_bid",
             EventKind::PayloadAttestationMessage(_) => "payload_attestation_message",
+            EventKind::FastConfirmation(_) => "fast_confirmation",
         }
     }
 
@@ -1396,6 +1404,11 @@ impl<E: EthSpec> EventKind<E> {
                     ))
                 })?,
             ))),
+            "fast_confirmation" => Ok(EventKind::FastConfirmation(
+                serde_json::from_str(data).map_err(|e| {
+                    ServerError::InvalidServerSentEvent(format!("Fast Confirmation: {:?}", e))
+                })?,
+            )),
             _ => Err(ServerError::InvalidServerSentEvent(
                 "Could not parse event tag".to_string(),
             )),
@@ -1436,6 +1449,7 @@ pub enum EventTopic {
     ExecutionPayloadAvailable,
     ExecutionPayloadBid,
     PayloadAttestationMessage,
+    FastConfirmation,
 }
 
 impl FromStr for EventTopic {
@@ -1466,6 +1480,7 @@ impl FromStr for EventTopic {
             "execution_payload_available" => Ok(EventTopic::ExecutionPayloadAvailable),
             "execution_payload_bid" => Ok(EventTopic::ExecutionPayloadBid),
             "payload_attestation_message" => Ok(EventTopic::PayloadAttestationMessage),
+            "fast_confirmation" => Ok(EventTopic::FastConfirmation),
             _ => Err("event topic cannot be parsed.".to_string()),
         }
     }
@@ -1501,6 +1516,7 @@ impl fmt::Display for EventTopic {
             EventTopic::PayloadAttestationMessage => {
                 write!(f, "payload_attestation_message")
             }
+            EventTopic::FastConfirmation => write!(f, "fast_confirmation"),
         }
     }
 }