Add Fast Confirmation Rule (FCR) implementation

Cargo.toml

@@ -100,7 +100,7 @@ unsafe_code = 'forbid'
 absolute_paths_not_starting_with_crate = 'warn'
 anonymous_parameters                   = 'warn'
 deprecated_in_future                   = 'warn'
-deprecated_safe                        = 'warn'
+deprecated_safe                        = { level = 'warn', priority = -1 }
 let_underscore_drop                    = 'warn'
 macro_use_extern_crate                 = 'warn'
 meta_variable_misuse                   = 'warn'

factory/src/lib.rs

@@ -167,6 +167,39 @@ pub fn block_justifying_current_epoch<P: Preset>(
     )
 }
 
+/// Creates a block at `block_slot` that includes attestations for the given `attestation_slots`.
+///
+/// Unlike [`block_justifying_current_epoch`], the caller controls both the block slot and the
+/// range of slots for which attestations are included. This is useful for tests that need to
+/// justify an epoch with a block that is not at the last slot of the epoch (for example, when
+/// the GU snapshot in FCR must be taken at a specific slot).
+pub fn block_with_attestations_for_slots<P: Preset>(
+    config: &Config,
+    pubkey_cache: &PubkeyCache,
+    pre_state: Arc<BeaconState<P>>,
+    block_slot: Slot,
+    attestation_slots: Range<Slot>,
+    graffiti: H256,
+) -> Result<BlockWithState<P>> {
+    let advanced_state = advance_state(config, pubkey_cache, pre_state, block_slot)?;
+    let eth1_data = advanced_state.eth1_data();
+    let attestations = full_block_attestations(config, &advanced_state, attestation_slots)?;
+    let deposits = ContiguousList::default();
+    let sync_aggregate = SyncAggregate::empty();
+
+    block(
+        config,
+        pubkey_cache,
+        advanced_state,
+        eth1_data,
+        graffiti,
+        attestations,
+        deposits,
+        sync_aggregate,
+        None,
+    )
+}
+
 pub fn block_with_deposits<P: Preset>(
     config: &Config,
     pubkey_cache: &PubkeyCache,

fork_choice_control/src/controller.rs

@@ -24,7 +24,7 @@ use execution_engine::{ExecutionEngine, PayloadStatusV1};
 use fork_choice_store::{
     AggregateAndProofOrigin, AttestationItem, AttestationOrigin, AttesterSlashingOrigin,
     BlobSidecarOrigin, BlockOrigin, DataColumnSidecarOrigin, ExecutionPayloadBidOrigin,
-    StateCacheProcessor, Store, StoreConfig,
+    FastConfirmationStore, StateCacheProcessor, Store, StoreConfig,
 };
 use futures::channel::{mpsc::Sender as MultiSender, oneshot::Sender as OneshotSender};
 use genesis::AnchorCheckpointProvider;
@@ -80,6 +80,8 @@ use crate::{
 pub struct Controller<P: Preset, E, A, W: Wait> {
     // The latest consistent snapshot of the store.
     store_snapshot: Arc<ArcSwap<Store<P, Storage<P>>>>,
+    // The latest consistent snapshot of the Fast Confirmation Rule store (None when FCR is disabled).
+    fcr_snapshot: Arc<ArcSwap<Option<FastConfirmationStore<P>>>>,
     block_processor: Arc<BlockProcessor<P>>,
     execution_engine: E,
     pubkey_cache: Arc<PubkeyCache>,
@@ -148,6 +150,17 @@ where
 
         store.apply_tick(tick)?;
 
+        // Instantiate the Fast Confirmation Rule store iff the feature is enabled. Per spec
+        // `get_fast_confirmation_store`, it anchors at `store.finalized_checkpoint`.
+        let fcr_store = store
+            .store_config()
+            .fast_confirmation_rule
+            .then(|| FastConfirmationStore::new(&store));
+        if let Some(metrics) = metrics.as_ref() {
+            metrics.set_beacon_fast_confirmation_enabled(fcr_store.is_some());
+        }
+        let fcr_snapshot = Arc::new(ArcSwap::from_pointee(fcr_store));
+
         let state_cache = store.state_cache();
         let store_snapshot = Arc::new(ArcSwap::from_pointee(store));
         let thread_pool = ThreadPool::new()?;
@@ -165,6 +178,7 @@ where
         let mut mutator = Mutator::new(
             pubkey_cache.clone_arc(),
             store_snapshot.clone_arc(),
+            fcr_snapshot.clone_arc(),
             state_cache.clone_arc(),
             block_processor.clone_arc(),
             event_channels,
@@ -199,6 +213,7 @@ where
 
         let controller = Arc::new(Self {
             store_snapshot,
+            fcr_snapshot,
             block_processor,
             execution_engine,
             pubkey_cache,
@@ -885,6 +900,10 @@ where
         self.store_snapshot.load_full()
     }
 
+    pub(crate) fn fcr_snapshot(&self) -> Guard<Arc<Option<FastConfirmationStore<P>>>> {
+        self.fcr_snapshot.load()
+    }
+
     pub(crate) fn owned_storage(&self) -> Arc<Storage<P>> {
         self.storage.clone_arc()
     }

fork_choice_control/src/events.rs

@@ -53,6 +53,7 @@ pub enum Topic {
     ContributionAndProof,
     DataColumnSidecar,
     ExecutionPayloadBid,
+    FastConfirmation,
     FinalizedCheckpoint,
     Head,
     PayloadAttributes,
@@ -73,6 +74,7 @@ pub enum Event<P: Preset> {
     ContributionAndProof(Box<SignedContributionAndProof<P>>),
     DataColumnSidecar(DataColumnSidecarEvent<P>),
     ExecutionPayloadBid(ExecutionPayloadBidEvent<P>),
+    FastConfirmation(FastConfirmationEvent),
     FinalizedCheckpoint(FinalizedCheckpointEvent),
     Head(HeadEvent),
     PayloadAttributes(PayloadAttributesEvent),
@@ -95,6 +97,7 @@ impl<P: Preset> Event<P> {
             Self::ContributionAndProof(_) => Topic::ContributionAndProof,
             Self::DataColumnSidecar(_) => Topic::DataColumnSidecar,
             Self::ExecutionPayloadBid(_) => Topic::ExecutionPayloadBid,
+            Self::FastConfirmation(_) => Topic::FastConfirmation,
             Self::FinalizedCheckpoint(_) => Topic::FinalizedCheckpoint,
             Self::Head(_) => Topic::Head,
             Self::PayloadAttributes(_) => Topic::PayloadAttributes,
@@ -118,6 +121,7 @@ pub struct EventChannels<P: Preset> {
     pub contribution_and_proofs: Sender<Event<P>>,
     pub data_column_sidecars: Sender<Event<P>>,
     pub execution_payload_bids: Sender<Event<P>>,
+    pub fast_confirmations: Sender<Event<P>>,
     pub finalized_checkpoints: Sender<Event<P>>,
     pub heads: Sender<Event<P>>,
     pub payload_attributes: Sender<Event<P>>,
@@ -148,6 +152,7 @@ impl<P: Preset> EventChannels<P> {
             contribution_and_proofs: broadcast::channel(max_events).0,
             data_column_sidecars: broadcast::channel(max_events).0,
             execution_payload_bids: broadcast::channel(max_events).0,
+            fast_confirmations: broadcast::channel(max_events).0,
             finalized_checkpoints: broadcast::channel(max_events).0,
             heads: broadcast::channel(max_events).0,
             payload_attributes: broadcast::channel(max_events).0,
@@ -171,6 +176,7 @@ impl<P: Preset> EventChannels<P> {
             Topic::ContributionAndProof => &self.contribution_and_proofs,
             Topic::DataColumnSidecar => &self.data_column_sidecars,
             Topic::ExecutionPayloadBid => &self.execution_payload_bids,
+            Topic::FastConfirmation => &self.fast_confirmations,
             Topic::FinalizedCheckpoint => &self.finalized_checkpoints,
             Topic::Head => &self.heads,
             Topic::PayloadAttributes => &self.payload_attributes,
@@ -275,6 +281,12 @@ impl<P: Preset> EventChannels<P> {
         }
     }
 
+    pub fn send_fast_confirmation_event(&self, block: H256, slot: Slot, current_slot: Slot) {
+        if let Err(error) = self.send_fast_confirmation_event_internal(block, slot, current_slot) {
+            warn_with_peers!("unable to send fast confirmation event: {error}");
+        }
+    }
+
     pub fn send_finalized_checkpoint_event(
         &self,
         block_root: H256,
@@ -504,6 +516,24 @@ impl<P: Preset> EventChannels<P> {
         Ok(())
     }
 
+    fn send_fast_confirmation_event_internal(
+        &self,
+        block: H256,
+        slot: Slot,
+        current_slot: Slot,
+    ) -> Result<()> {
+        if self.fast_confirmations.receiver_count() > 0 {
+            let event = Event::FastConfirmation(FastConfirmationEvent {
+                block,
+                slot,
+                current_slot,
+            });
+            self.fast_confirmations.send(event)?;
+        }
+
+        Ok(())
+    }
+
     fn send_finalized_checkpoint_event_internal(
         &self,
         block_root: H256,
@@ -655,6 +685,27 @@ pub struct BlockGossipEvent {
     pub block: H256,
 }
 
+/// SSE event payload for the `fast_confirmation` topic ([beacon-APIs PR #598], extended by
+/// [beacon-APIs PR #616]).
+///
+/// Emitted from `mutator.rs` once per slot, after `FastConfirmationStore::on_fast_confirmation`
+/// runs — regardless of whether the most recent confirmed block changed since the previous run.
+/// `current_slot` lets consumers distinguish a fresh re-confirmation of the same root from a
+/// stale stream.
+///
+/// JSON wire form: `{"block": "0x...", "slot": "N", "current_slot": "M"}`.
+///
+/// [beacon-APIs PR #598]: https://github.com/ethereum/beacon-APIs/pull/598
+/// [beacon-APIs PR #616]: https://github.com/ethereum/beacon-APIs/pull/616
+#[derive(Clone, Copy, Debug, Serialize)]
+pub struct FastConfirmationEvent {
+    pub block: H256,
+    #[serde(with = "serde_utils::string_or_native")]
+    pub slot: Slot,
+    #[serde(with = "serde_utils::string_or_native")]
+    pub current_slot: Slot,
+}
+
 #[derive(Clone, Debug, Serialize)]
 pub struct DataColumnSidecarEvent<P: Preset> {
     pub block_root: H256,