Enable proposer boost re-orging (#2860)

## Proposed Changes

With proposer boosting implemented (#2822) we have an opportunity to re-org out late blocks.

This PR adds three flags to the BN to control this behaviour:

* `--disable-proposer-reorgs`: turn aggressive re-orging off (it's on by default).
* `--proposer-reorg-threshold N`: attempt to orphan blocks with less than N% of the committee vote. If this parameter isn't set then N defaults to 20% when the feature is enabled.
* `--proposer-reorg-epochs-since-finalization N`: only attempt to re-org late blocks when the number of epochs since finalization is less than or equal to N. The default is 2 epochs, meaning re-orgs will only be attempted when the chain is finalizing optimally.

For safety Lighthouse will only attempt a re-org under very specific conditions:

1. The block being proposed is 1 slot after the canonical head, and the canonical head is 1 slot after its parent. i.e. at slot `n + 1` rather than building on the block from slot `n` we build on the block from slot `n - 1`.
2. The current canonical head received less than N% of the committee vote. N should be set depending on the proposer boost fraction itself, the fraction of the network that is believed to be applying it, and the size of the largest entity that could be hoarding votes.
3. The current canonical head arrived after the attestation deadline from our perspective. This condition was only added to support suppression of forkchoiceUpdated messages, but makes intuitive sense.
4. The block is being proposed in the first 2 seconds of the slot. This gives it time to propagate and receive the proposer boost.


## Additional Info

For the initial idea and background, see: ethereum/consensus-specs#2353 (comment)

There is also a specification for this feature here: ethereum/consensus-specs#3034

Co-authored-by: Michael Sproul <micsproul@gmail.com>
Co-authored-by: pawan <pawandhananjay@gmail.com>

.github/custom/clippy.toml

@@ -16,6 +16,7 @@ async-wrapper-methods = [
     "task_executor::TaskExecutor::spawn_blocking_handle",
     "warp_utils::task::blocking_task",
     "warp_utils::task::blocking_json_task",
+    "beacon_chain::beacon_chain::BeaconChain::spawn_blocking_handle",
     "validator_client::http_api::blocking_signed_json_task",
     "execution_layer::test_utils::MockServer::new",
     "execution_layer::test_utils::MockServer::new_with_config",

beacon_node/beacon_chain/src/beacon_fork_choice_store.rs

@@ -7,6 +7,8 @@
 use crate::{metrics, BeaconSnapshot};
 use derivative::Derivative;
 use fork_choice::ForkChoiceStore;
+use proto_array::JustifiedBalances;
+use safe_arith::ArithError;
 use ssz_derive::{Decode, Encode};
 use std::collections::BTreeSet;
 use std::marker::PhantomData;
@@ -31,6 +33,7 @@ pub enum Error {
     MissingState(Hash256),
     InvalidPersistedBytes(ssz::DecodeError),
     BeaconStateError(BeaconStateError),
+    Arith(ArithError),
 }
 
 impl From<BeaconStateError> for Error {
@@ -39,27 +42,15 @@ impl From<BeaconStateError> for Error {
     }
 }
 
+impl From<ArithError> for Error {
+    fn from(e: ArithError) -> Self {
+        Error::Arith(e)
+    }
+}
+
 /// The number of validator balance sets that are cached within `BalancesCache`.
 const MAX_BALANCE_CACHE_SIZE: usize = 4;
 
-/// Returns the effective balances for every validator in the given `state`.
-///
-/// Any validator who is not active in the epoch of the given `state` is assigned a balance of
-/// zero.
-pub fn get_effective_balances<T: EthSpec>(state: &BeaconState<T>) -> Vec<u64> {
-    state
-        .validators()
-        .iter()
-        .map(|validator| {
-            if validator.is_active_at(state.current_epoch()) {
-                validator.effective_balance
-            } else {
-                0
-            }
-        })
-        .collect()
-}
-
 #[superstruct(
     variants(V8),
     variant_attributes(derive(PartialEq, Clone, Debug, Encode, Decode)),
@@ -113,7 +104,7 @@ impl BalancesCache {
             let item = CacheItem {
                 block_root: epoch_boundary_root,
                 epoch,
-                balances: get_effective_balances(state),
+                balances: JustifiedBalances::from_justified_state(state)?.effective_balances,
             };
 
             if self.items.len() == MAX_BALANCE_CACHE_SIZE {
@@ -152,7 +143,7 @@ pub struct BeaconForkChoiceStore<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<
     time: Slot,
     finalized_checkpoint: Checkpoint,
     justified_checkpoint: Checkpoint,
-    justified_balances: Vec<u64>,
+    justified_balances: JustifiedBalances,
     best_justified_checkpoint: Checkpoint,
     unrealized_justified_checkpoint: Checkpoint,
     unrealized_finalized_checkpoint: Checkpoint,
@@ -181,7 +172,7 @@ where
     pub fn get_forkchoice_store(
         store: Arc<HotColdDB<E, Hot, Cold>>,
         anchor: &BeaconSnapshot<E>,
-    ) -> Self {
+    ) -> Result<Self, Error> {
         let anchor_state = &anchor.beacon_state;
         let mut anchor_block_header = anchor_state.latest_block_header().clone();
         if anchor_block_header.state_root == Hash256::zero() {
@@ -194,21 +185,22 @@ where
             root: anchor_root,
         };
         let finalized_checkpoint = justified_checkpoint;
+        let justified_balances = JustifiedBalances::from_justified_state(anchor_state)?;
 
-        Self {
+        Ok(Self {
             store,
             balances_cache: <_>::default(),
             time: anchor_state.slot(),
             justified_checkpoint,
-            justified_balances: anchor_state.balances().clone().into(),
+            justified_balances,
             finalized_checkpoint,
             best_justified_checkpoint: justified_checkpoint,
             unrealized_justified_checkpoint: justified_checkpoint,
             unrealized_finalized_checkpoint: finalized_checkpoint,
             proposer_boost_root: Hash256::zero(),
             equivocating_indices: BTreeSet::new(),
             _phantom: PhantomData,
-        }
+        })
     }
 
     /// Save the current state of `Self` to a `PersistedForkChoiceStore` which can be stored to the
@@ -219,7 +211,7 @@ where
             time: self.time,
             finalized_checkpoint: self.finalized_checkpoint,
             justified_checkpoint: self.justified_checkpoint,
-            justified_balances: self.justified_balances.clone(),
+            justified_balances: self.justified_balances.effective_balances.clone(),
             best_justified_checkpoint: self.best_justified_checkpoint,
             unrealized_justified_checkpoint: self.unrealized_justified_checkpoint,
             unrealized_finalized_checkpoint: self.unrealized_finalized_checkpoint,
@@ -233,13 +225,15 @@ where
         persisted: PersistedForkChoiceStore,
         store: Arc<HotColdDB<E, Hot, Cold>>,
     ) -> Result<Self, Error> {
+        let justified_balances =
+            JustifiedBalances::from_effective_balances(persisted.justified_balances)?;
         Ok(Self {
             store,
             balances_cache: persisted.balances_cache,
             time: persisted.time,
             finalized_checkpoint: persisted.finalized_checkpoint,
             justified_checkpoint: persisted.justified_checkpoint,
-            justified_balances: persisted.justified_balances,
+            justified_balances,
             best_justified_checkpoint: persisted.best_justified_checkpoint,
             unrealized_justified_checkpoint: persisted.unrealized_justified_checkpoint,
             unrealized_finalized_checkpoint: persisted.unrealized_finalized_checkpoint,
@@ -279,7 +273,7 @@ where
         &self.justified_checkpoint
     }
 
-    fn justified_balances(&self) -> &[u64] {
+    fn justified_balances(&self) -> &JustifiedBalances {
         &self.justified_balances
     }
 
@@ -314,8 +308,9 @@ where
             self.justified_checkpoint.root,
             self.justified_checkpoint.epoch,
         ) {
+            // NOTE: could avoid this re-calculation by introducing a `PersistedCacheItem`.
             metrics::inc_counter(&metrics::BALANCES_CACHE_HITS);
-            self.justified_balances = balances;
+            self.justified_balances = JustifiedBalances::from_effective_balances(balances)?;
         } else {
             metrics::inc_counter(&metrics::BALANCES_CACHE_MISSES);
             let justified_block = self
@@ -332,7 +327,7 @@ where
                 .map_err(Error::FailedToReadState)?
                 .ok_or_else(|| Error::MissingState(justified_block.state_root()))?;
 
-            self.justified_balances = get_effective_balances(&state);
+            self.justified_balances = JustifiedBalances::from_justified_state(&state)?;
         }
 
         Ok(())

beacon_node/beacon_chain/src/builder.rs

@@ -22,6 +22,7 @@ use fork_choice::{ForkChoice, ResetPayloadStatuses};
 use futures::channel::mpsc::Sender;
 use operation_pool::{OperationPool, PersistedOperationPool};
 use parking_lot::RwLock;
+use proto_array::ReOrgThreshold;
 use slasher::Slasher;
 use slog::{crit, error, info, Logger};
 use slot_clock::{SlotClock, TestingSlotClock};
@@ -31,8 +32,8 @@ use std::time::Duration;
 use store::{Error as StoreError, HotColdDB, ItemStore, KeyValueStoreOp};
 use task_executor::{ShutdownReason, TaskExecutor};
 use types::{
-    BeaconBlock, BeaconState, ChainSpec, Checkpoint, EthSpec, Graffiti, Hash256, PublicKeyBytes,
-    Signature, SignedBeaconBlock, Slot,
+    BeaconBlock, BeaconState, ChainSpec, Checkpoint, Epoch, EthSpec, Graffiti, Hash256,
+    PublicKeyBytes, Signature, SignedBeaconBlock, Slot,
 };
 
 /// An empty struct used to "witness" all the `BeaconChainTypes` traits. It has no user-facing
@@ -159,6 +160,21 @@ where
         self
     }
 
+    /// Sets the proposer re-org threshold.
+    pub fn proposer_re_org_threshold(mut self, threshold: Option<ReOrgThreshold>) -> Self {
+        self.chain_config.re_org_threshold = threshold;
+        self
+    }
+
+    /// Sets the proposer re-org max epochs since finalization.
+    pub fn proposer_re_org_max_epochs_since_finalization(
+        mut self,
+        epochs_since_finalization: Epoch,
+    ) -> Self {
+        self.chain_config.re_org_max_epochs_since_finalization = epochs_since_finalization;
+        self
+    }
+
     /// Sets the store (database).
     ///
     /// Should generally be called early in the build chain.
@@ -358,7 +374,8 @@ where
         let (genesis, updated_builder) = self.set_genesis_state(beacon_state)?;
         self = updated_builder;
 
-        let fc_store = BeaconForkChoiceStore::get_forkchoice_store(store, &genesis);
+        let fc_store = BeaconForkChoiceStore::get_forkchoice_store(store, &genesis)
+            .map_err(|e| format!("Unable to initialize fork choice store: {e:?}"))?;
         let current_slot = None;
 
         let fork_choice = ForkChoice::from_anchor(
@@ -476,7 +493,8 @@ where
             beacon_state: weak_subj_state,
         };
 
-        let fc_store = BeaconForkChoiceStore::get_forkchoice_store(store, &snapshot);
+        let fc_store = BeaconForkChoiceStore::get_forkchoice_store(store, &snapshot)
+            .map_err(|e| format!("Unable to initialize fork choice store: {e:?}"))?;
 
         let current_slot = Some(snapshot.beacon_block.slot());
         let fork_choice = ForkChoice::from_anchor(

beacon_node/beacon_chain/src/canonical_head.rs

@@ -34,7 +34,8 @@
 use crate::persisted_fork_choice::PersistedForkChoice;
 use crate::{
     beacon_chain::{
-        BeaconForkChoice, BeaconStore, BLOCK_PROCESSING_CACHE_LOCK_TIMEOUT, FORK_CHOICE_DB_KEY,
+        BeaconForkChoice, BeaconStore, OverrideForkchoiceUpdate,
+        BLOCK_PROCESSING_CACHE_LOCK_TIMEOUT, FORK_CHOICE_DB_KEY,
     },
     block_times_cache::BlockTimesCache,
     events::ServerSentEventHandler,
@@ -114,6 +115,11 @@ impl<E: EthSpec> CachedHead<E> {
         self.snapshot.beacon_block_root
     }
 
+    /// Returns the root of the parent of the head block.
+    pub fn parent_block_root(&self) -> Hash256 {
+        self.snapshot.beacon_block.parent_root()
+    }
+
     /// Returns root of the `BeaconState` at the head of the beacon chain.
     ///
     /// ## Note
@@ -146,6 +152,21 @@ impl<E: EthSpec> CachedHead<E> {
         Ok(root)
     }
 
+    /// Returns the randao mix for the parent of the block at the head of the chain.
+    ///
+    /// This is useful for re-orging the current head. The parent's RANDAO value is read from
+    /// the head's execution payload because it is unavailable in the beacon state's RANDAO mixes
+    /// array after being overwritten by the head block's RANDAO mix.
+    ///
+    /// This will error if the head block is not execution-enabled (post Bellatrix).
+    pub fn parent_random(&self) -> Result<Hash256, BeaconStateError> {
+        self.snapshot
+            .beacon_block
+            .message()
+            .execution_payload()
+            .map(|payload| payload.prev_randao())
+    }
+
     /// Returns the active validator count for the current epoch of the head state.
     ///
     /// Should only return `None` if the caches have not been built on the head state (this should
@@ -765,6 +786,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         new_cached_head: &CachedHead<T::EthSpec>,
         new_head_proto_block: ProtoBlock,
     ) -> Result<(), Error> {
+        let _timer = metrics::start_timer(&metrics::FORK_CHOICE_AFTER_NEW_HEAD_TIMES);
         let old_snapshot = &old_cached_head.snapshot;
         let new_snapshot = &new_cached_head.snapshot;
         let new_head_is_optimistic = new_head_proto_block
@@ -902,6 +924,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         new_view: ForkChoiceView,
         finalized_proto_block: ProtoBlock,
     ) -> Result<(), Error> {
+        let _timer = metrics::start_timer(&metrics::FORK_CHOICE_AFTER_FINALIZATION_TIMES);
         let new_snapshot = &new_cached_head.snapshot;
         let finalized_block_is_optimistic = finalized_proto_block
             .execution_status
@@ -1124,7 +1147,11 @@ fn spawn_execution_layer_updates<T: BeaconChainTypes>(
                 }
 
                 if let Err(e) = chain
-                    .update_execution_engine_forkchoice(current_slot, forkchoice_update_params)
+                    .update_execution_engine_forkchoice(
+                        current_slot,
+                        forkchoice_update_params,
+                        OverrideForkchoiceUpdate::Yes,
+                    )
                     .await
                 {
                     crit!(

beacon_node/beacon_chain/src/chain_config.rs

@@ -1,9 +1,18 @@
-pub use proto_array::CountUnrealizedFull;
+pub use proto_array::{CountUnrealizedFull, ReOrgThreshold};
 use serde_derive::{Deserialize, Serialize};
-use types::Checkpoint;
+use std::time::Duration;
+use types::{Checkpoint, Epoch};
 
+pub const DEFAULT_RE_ORG_THRESHOLD: ReOrgThreshold = ReOrgThreshold(20);
+pub const DEFAULT_RE_ORG_MAX_EPOCHS_SINCE_FINALIZATION: Epoch = Epoch::new(2);
 pub const DEFAULT_FORK_CHOICE_BEFORE_PROPOSAL_TIMEOUT: u64 = 250;
 
+/// Default fraction of a slot lookahead for payload preparation (12/3 = 4 seconds on mainnet).
+pub const DEFAULT_PREPARE_PAYLOAD_LOOKAHEAD_FACTOR: u32 = 3;
+
+/// Fraction of a slot lookahead for fork choice in the state advance timer (500ms on mainnet).
+pub const FORK_CHOICE_LOOKAHEAD_FACTOR: u32 = 24;
+
 #[derive(Debug, PartialEq, Eq, Clone, Deserialize, Serialize)]
 pub struct ChainConfig {
     /// Maximum number of slots to skip when importing a consensus message (e.g., block,
@@ -21,6 +30,10 @@ pub struct ChainConfig {
     pub enable_lock_timeouts: bool,
     /// The max size of a message that can be sent over the network.
     pub max_network_size: usize,
+    /// Maximum percentage of committee weight at which to attempt re-orging the canonical head.
+    pub re_org_threshold: Option<ReOrgThreshold>,
+    /// Maximum number of epochs since finalization for attempting a proposer re-org.
+    pub re_org_max_epochs_since_finalization: Epoch,
     /// Number of milliseconds to wait for fork choice before proposing a block.
     ///
     /// If set to 0 then block proposal will not wait for fork choice at all.
@@ -47,6 +60,11 @@ pub struct ChainConfig {
     pub count_unrealized_full: CountUnrealizedFull,
     /// Optionally set timeout for calls to checkpoint sync endpoint.
     pub checkpoint_sync_url_timeout: u64,
+    /// The offset before the start of a proposal slot at which payload attributes should be sent.
+    ///
+    /// Low values are useful for execution engines which don't improve their payload after the
+    /// first call, and high values are useful for ensuring the EL is given ample notice.
+    pub prepare_payload_lookahead: Duration,
 }
 
 impl Default for ChainConfig {
@@ -57,6 +75,8 @@ impl Default for ChainConfig {
             reconstruct_historic_states: false,
             enable_lock_timeouts: true,
             max_network_size: 10 * 1_048_576, // 10M
+            re_org_threshold: Some(DEFAULT_RE_ORG_THRESHOLD),
+            re_org_max_epochs_since_finalization: DEFAULT_RE_ORG_MAX_EPOCHS_SINCE_FINALIZATION,
             fork_choice_before_proposal_timeout_ms: DEFAULT_FORK_CHOICE_BEFORE_PROPOSAL_TIMEOUT,
             // Builder fallback configs that are set in `clap` will override these.
             builder_fallback_skips: 3,
@@ -68,6 +88,7 @@ impl Default for ChainConfig {
             paranoid_block_proposal: false,
             count_unrealized_full: CountUnrealizedFull::default(),
             checkpoint_sync_url_timeout: 60,
+            prepare_payload_lookahead: Duration::from_secs(4),
         }
     }
 }

beacon_node/beacon_chain/src/errors.rs

@@ -204,6 +204,7 @@ pub enum BeaconChainError {
     MissingPersistedForkChoice,
     CommitteePromiseFailed(oneshot_broadcast::Error),
     MaxCommitteePromises(usize),
+    ProposerHeadForkChoiceError(fork_choice::Error<proto_array::Error>),
 }
 
 easy_from_to!(SlotProcessingError, BeaconChainError);
@@ -234,6 +235,7 @@ pub enum BlockProductionError {
     UnableToProduceAtSlot(Slot),
     SlotProcessingError(SlotProcessingError),
     BlockProcessingError(BlockProcessingError),
+    ForkChoiceError(ForkChoiceError),
     Eth1ChainError(Eth1ChainError),
     BeaconStateError(BeaconStateError),
     StateAdvanceError(StateAdvanceError),
@@ -252,7 +254,6 @@ pub enum BlockProductionError {
     FailedToReadFinalizedBlock(store::Error),
     MissingFinalizedBlock(Hash256),
     BlockTooLarge(usize),
-    ForkChoiceError(BeaconChainError),
     ShuttingDown,
     MissingSyncAggregate,
     MissingExecutionPayload,
@@ -265,3 +266,4 @@ easy_from_to!(BeaconStateError, BlockProductionError);
 easy_from_to!(SlotProcessingError, BlockProductionError);
 easy_from_to!(Eth1ChainError, BlockProductionError);
 easy_from_to!(StateAdvanceError, BlockProductionError);
+easy_from_to!(ForkChoiceError, BlockProductionError);