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

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -36,7 +36,7 @@ use crate::persisted_beacon_chain::{PersistedBeaconChain, DUMMY_CANONICAL_HEAD_B
 use crate::persisted_fork_choice::PersistedForkChoice;
 use crate::pre_finalization_cache::PreFinalizationBlockCache;
 use crate::shuffling_cache::{BlockShufflingIds, ShufflingCache};
-use crate::snapshot_cache::SnapshotCache;
+use crate::snapshot_cache::{BlockProductionPreState, SnapshotCache};
 use crate::sync_committee_verification::{
     Error as SyncCommitteeError, VerifiedSyncCommitteeMessage, VerifiedSyncContribution,
 };
@@ -103,6 +103,12 @@ pub const ATTESTATION_CACHE_LOCK_TIMEOUT: Duration = Duration::from_secs(1);
 /// validator pubkey cache.
 pub const VALIDATOR_PUBKEY_CACHE_LOCK_TIMEOUT: Duration = Duration::from_secs(1);
 
+/// The latest delay from the start of the slot at which to attempt a 1-slot re-org.
+fn max_re_org_slot_delay(seconds_per_slot: u64) -> Duration {
+    // Allow at least half of the attestation deadline for the block to propagate.
+    Duration::from_secs(seconds_per_slot) / 6
+}
+
 // These keys are all zero because they get stored in different columns, see `DBColumn` type.
 pub const BEACON_CHAIN_DB_KEY: Hash256 = Hash256::zero();
 pub const OP_POOL_DB_KEY: Hash256 = Hash256::zero();
@@ -2892,8 +2898,18 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             .head_info()
             .map_err(BlockProductionError::UnableToGetHeadInfo)?;
         let (state, state_root_opt) = if head_info.slot < slot {
+            // Attempt an aggressive re-org if configured and the conditions are right.
+            if let Some(re_org_state) = self.get_state_for_re_org(slot, &head_info)? {
+                info!(
+                    self.log,
+                    "Proposing block to re-org current head";
+                    "slot" => slot,
+                    "head" => %head_info.block_root,
+                );
+                (re_org_state.pre_state, re_org_state.state_root)
+            }
             // Normal case: proposing a block atop the current head. Use the snapshot cache.
-            if let Some(pre_state) = self
+            else if let Some(pre_state) = self
                 .snapshot_cache
                 .try_read_for(BLOCK_PROCESSING_CACHE_LOCK_TIMEOUT)
                 .and_then(|snapshot_cache| {
@@ -2938,6 +2954,87 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         )
     }
 
+    fn get_state_for_re_org(
+        &self,
+        slot: Slot,
+        head_info: &HeadInfo,
+    ) -> Result<Option<BlockProductionPreState<T::EthSpec>>, BlockProductionError> {
+        if let Some(re_org_threshold) = self.config.re_org_threshold {
+            if self.spec.proposer_score_boost.is_none() {
+                warn!(
+                    self.log,
+                    "Ignoring proposer re-org configuration";
+                    "reason" => "this network does not have proposer boosting enabled"
+                );
+                return Ok(None);
+            }
+
+            let canonical_head = head_info.block_root;
+            let slot_delay = self
+                .slot_clock
+                .seconds_from_current_slot_start(self.spec.seconds_per_slot)
+                .ok_or(BlockProductionError::UnableToReadSlot)?;
+
+            // Check that we're producing a block one slot after the current head, and early enough
+            // in the slot to be able to propagate widely.
+            if head_info.slot + 1 == slot
+                && slot_delay < max_re_org_slot_delay(self.spec.seconds_per_slot)
+            {
+                // Is the current head weak and appropriate for re-orging?
+                let proposer_head = self.fork_choice.write().get_proposer_head(
+                    slot,
+                    canonical_head,
+                    re_org_threshold,
+                )?;
+                if let Some(re_org_head) = proposer_head.re_org_head {
+                    // Only attempt a re-org if we hit the snapshot cache.
+                    if let Some(pre_state) = self
+                        .snapshot_cache
+                        .try_read_for(BLOCK_PROCESSING_CACHE_LOCK_TIMEOUT)
+                        .and_then(|snapshot_cache| {
+                            snapshot_cache.get_state_for_block_production(re_org_head)
+                        })
+                    {
+                        debug!(
+                            self.log,
+                            "Attempting re-org due to weak head";
+                            "head" => ?canonical_head,
+                            "re_org_head" => ?re_org_head,
+                            "head_weight" => ?proposer_head.canonical_head_weight,
+                            "re_org_weight" => ?proposer_head.re_org_weight_threshold,
+                        );
+                        return Ok(Some(pre_state));
+                    } else {
+                        debug!(
+                            self.log,
+                            "Not attempting re-org due to cache miss";
+                            "head" => ?canonical_head,
+                            "re_org_head" => ?re_org_head,
+                            "head_weight" => ?proposer_head.canonical_head_weight,
+                            "re_org_weight" => ?proposer_head.re_org_weight_threshold,
+                        );
+                    }
+                } else {
+                    debug!(
+                        self.log,
+                        "Not attempting re-org due to strong head";
+                        "head" => ?canonical_head,
+                        "head_weight" => ?proposer_head.canonical_head_weight,
+                        "re_org_weight" => ?proposer_head.re_org_weight_threshold,
+                    );
+                }
+            } else {
+                debug!(
+                    self.log,
+                    "Not attempting re-org due to slot distance";
+                    "head" => ?canonical_head,
+                );
+            }
+        }
+
+        Ok(None)
+    }
+
     /// Produce a block for some `slot` upon the given `state`.
     ///
     /// Typically the `self.produce_block()` function should be used, instead of calling this

beacon_node/beacon_chain/src/chain_config.rs

@@ -1,6 +1,8 @@
 use serde_derive::{Deserialize, Serialize};
 use types::Checkpoint;
 
+pub const DEFAULT_RE_ORG_THRESHOLD: u64 = 10;
+
 #[derive(Debug, PartialEq, Eq, Clone, Deserialize, Serialize)]
 pub struct ChainConfig {
     /// Maximum number of slots to skip when importing a consensus message (e.g., block,
@@ -18,6 +20,8 @@ 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 weight at which to attempt re-orging the canonical head.
+    pub re_org_threshold: Option<u64>,
 }
 
 impl Default for ChainConfig {
@@ -28,6 +32,7 @@ impl Default for ChainConfig {
             reconstruct_historic_states: false,
             enable_lock_timeouts: true,
             max_network_size: 10 * 1_048_576, // 10M
+            re_org_threshold: None,
         }
     }
 }

beacon_node/beacon_chain/src/errors.rs

@@ -191,6 +191,7 @@ pub enum BlockProductionError {
     UnableToProduceAtSlot(Slot),
     SlotProcessingError(SlotProcessingError),
     BlockProcessingError(BlockProcessingError),
+    ForkChoiceError(ForkChoiceError),
     Eth1ChainError(Eth1ChainError),
     BeaconStateError(BeaconStateError),
     StateAdvanceError(StateAdvanceError),
@@ -216,3 +217,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);

beacon_node/beacon_chain/src/snapshot_cache.rs

@@ -12,7 +12,10 @@ pub const DEFAULT_SNAPSHOT_CACHE_SIZE: usize = 4;
 
 /// The minimum block delay to clone the state in the cache instead of removing it.
 /// This helps keep block processing fast during re-orgs from late blocks.
-const MINIMUM_BLOCK_DELAY_FOR_CLONE: Duration = Duration::from_secs(6);
+fn minimum_block_delay_for_clone(seconds_per_slot: u64) -> Duration {
+    // If the block arrived at the attestation deadline or later, it might get re-orged.
+    Duration::from_secs(seconds_per_slot) / 3
+}
 
 /// This snapshot is to be used for verifying a child of `self.beacon_block`.
 #[derive(Debug)]
@@ -257,7 +260,7 @@ impl<T: EthSpec> SnapshotCache<T> {
                         return (cache.clone_as_pre_state(), true);
                     }
                     if let Some(delay) = block_delay {
-                        if delay >= MINIMUM_BLOCK_DELAY_FOR_CLONE
+                        if delay >= minimum_block_delay_for_clone(spec.seconds_per_slot)
                             && delay <= Duration::from_secs(spec.seconds_per_slot) * 4
                         {
                             return (cache.clone_as_pre_state(), true);

beacon_node/src/cli.rs

@@ -650,4 +650,18 @@ pub fn cli_app<'a, 'b>() -> App<'a, 'b> {
                        experimental as it may obscure performance issues.")
                 .takes_value(false)
         )
+        .arg(
+            Arg::with_name("enable-proposer-re-orgs")
+                .long("enable-proposer-re-orgs")
+                .help("Attempt to re-org out weak/late blocks from other proposers \
+                       (dangerous, experimental)")
+                .takes_value(true)
+        )
+        .arg(
+            Arg::with_name("proposer-re-org-fraction")
+                .long("proposer-re-org-fraction")
+                .help("Percentage of vote weight below which to attempt a proposer re-org")
+                .requires("enable-proposer-re-orgs")
+                .takes_value(true)
+        )
 }

beacon_node/src/config.rs

@@ -1,3 +1,4 @@
+use beacon_chain::chain_config::DEFAULT_RE_ORG_THRESHOLD;
 use clap::ArgMatches;
 use clap_utils::flags::DISABLE_MALLOC_TUNING_FLAG;
 use client::{ClientConfig, ClientGenesis};
@@ -559,6 +560,17 @@ pub fn get_config<E: EthSpec>(
         client_config.chain.enable_lock_timeouts = false;
     }
 
+    if let Some(enable_re_orgs) = clap_utils::parse_optional(cli_args, "enable-proposer-re-orgs")? {
+        if enable_re_orgs {
+            client_config.chain.re_org_threshold = Some(
+                clap_utils::parse_optional(cli_args, "proposer-re-org-fraction")?
+                    .unwrap_or(DEFAULT_RE_ORG_THRESHOLD),
+            );
+        } else {
+            client_config.chain.re_org_threshold = None;
+        }
+    }
+
     Ok(client_config)
 }
 

consensus/fork_choice/src/fork_choice.rs

@@ -1,5 +1,5 @@
 use crate::{ForkChoiceStore, InvalidationOperation};
-use proto_array::{Block as ProtoBlock, ExecutionStatus, ProtoArrayForkChoice};
+use proto_array::{Block as ProtoBlock, ExecutionStatus, ProposerHead, ProtoArrayForkChoice};
 use ssz_derive::{Decode, Encode};
 use std::cmp::Ordering;
 use std::marker::PhantomData;
@@ -425,6 +425,25 @@ where
             .map_err(Into::into)
     }
 
+    pub fn get_proposer_head(
+        &mut self,
+        current_slot: Slot,
+        canonical_head: Hash256,
+        re_org_threshold: u64,
+    ) -> Result<ProposerHead, Error<T::Error>> {
+        // Calling `update_time` is essential, as it needs to dequeue attestations from the previous
+        // slot so we can see how many attesters voted for the canonical head.
+        self.update_time(current_slot)?;
+
+        self.proto_array
+            .get_proposer_head::<E>(
+                self.fc_store.justified_balances(),
+                canonical_head,
+                re_org_threshold,
+            )
+            .map_err(Into::into)
+    }
+
     /// Returns `true` if the given `store` should be updated to set
     /// `state.current_justified_checkpoint` its `justified_checkpoint`.
     ///

consensus/proto_array/src/error.rs

@@ -15,6 +15,7 @@ pub enum Error {
     InvalidNodeDelta(usize),
     DeltaOverflow(usize),
     ProposerBoostOverflow(usize),
+    UniqueWeightOverflow(Hash256),
     IndexOverflow(&'static str),
     InvalidExecutionDeltaOverflow(usize),
     InvalidDeltaLen {

consensus/proto_array/src/lib.rs

@@ -5,7 +5,9 @@ mod proto_array_fork_choice;
 mod ssz_container;
 
 pub use crate::proto_array::InvalidationOperation;
-pub use crate::proto_array_fork_choice::{Block, ExecutionStatus, ProtoArrayForkChoice};
+pub use crate::proto_array_fork_choice::{
+    Block, ExecutionStatus, ProposerHead, ProtoArrayForkChoice,
+};
 pub use error::Error;
 
 pub mod core {

consensus/proto_array/src/proto_array.rs

@@ -898,7 +898,7 @@ impl ProtoArray {
 /// Returns `None` if there is an overflow or underflow when calculating the score.
 ///
 /// https://github.com/ethereum/consensus-specs/blob/dev/specs/phase0/fork-choice.md#get_latest_attesting_balance
-fn calculate_proposer_boost<E: EthSpec>(
+pub fn calculate_proposer_boost<E: EthSpec>(
     validator_balances: &[u64],
     proposer_score_boost: u64,
 ) -> Option<u64> {