[Merged by Bors] - Enable proposer boost re-orging

.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);