Fix bugs in proposer calculation post-Fulu (#8101)

As identified by a researcher during the Fusaka security competition, we were computing the proposer index incorrectly in some places by computing without lookahead.


  - [x] Add "low level" checks to computation functions in `consensus/types` to ensure they error cleanly
- [x] Re-work the determination of proposer shuffling decision roots, which are now fork aware.
- [x] Re-work and simplify the beacon proposer cache to be fork-aware.
- [x] Optimise `with_proposer_cache` to use `OnceCell`.
- [x] All tests passing.
- [x] Resolve all remaining `FIXME(sproul)`s.
- [x] Unit tests for `ProtoBlock::proposer_shuffling_root_for_child_block`.
- [x] End-to-end regression test.
- [x] Test on pre-Fulu network.
- [x] Test on post-Fulu network.


Co-Authored-By: Michael Sproul <michael@sigmaprime.io>

Author: michaelsproul

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -5,8 +5,9 @@ use crate::attestation_verification::{
 };
 use crate::attester_cache::{AttesterCache, AttesterCacheKey};
 use crate::beacon_block_streamer::{BeaconBlockStreamer, CheckCaches};
-use crate::beacon_proposer_cache::BeaconProposerCache;
-use crate::beacon_proposer_cache::compute_proposer_duties_from_head;
+use crate::beacon_proposer_cache::{
+    BeaconProposerCache, EpochBlockProposers, ensure_state_can_determine_proposers_for_epoch,
+};
 use crate::blob_verification::{GossipBlobError, GossipVerifiedBlob};
 use crate::block_times_cache::BlockTimesCache;
 use crate::block_verification::POS_PANDA_BANNER;
@@ -4698,65 +4699,54 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
 
         // Compute the proposer index.
         let head_epoch = cached_head.head_slot().epoch(T::EthSpec::slots_per_epoch());
-        let shuffling_decision_root = if head_epoch == proposal_epoch {
-            cached_head
-                .snapshot
-                .beacon_state
-                .proposer_shuffling_decision_root(proposer_head)?
-        } else {
-            proposer_head
-        };
-        let cached_proposer = self
-            .beacon_proposer_cache
-            .lock()
-            .get_slot::<T::EthSpec>(shuffling_decision_root, proposal_slot);
-        let proposer_index = if let Some(proposer) = cached_proposer {
-            proposer.index as u64
-        } else {
-            if head_epoch + self.config.sync_tolerance_epochs < proposal_epoch {
-                warn!(
-                    msg = "this is a non-critical issue that can happen on unhealthy nodes or \
-                              networks.",
-                    %proposal_epoch,
-                    %head_epoch,
-                    "Skipping proposer preparation"
-                );
-
-                // Don't skip the head forward more than two epochs. This avoids burdening an
-                // unhealthy node.
-                //
-                // Although this node might miss out on preparing for a proposal, they should still
-                // be able to propose. This will prioritise beacon chain health over efficient
-                // packing of execution blocks.
-                return Ok(None);
-            }
-
-            let (proposers, decision_root, _, fork) =
-                compute_proposer_duties_from_head(proposal_epoch, self)?;
-
-            let proposer_offset = (proposal_slot % T::EthSpec::slots_per_epoch()).as_usize();
-            let proposer = *proposers
-                .get(proposer_offset)
-                .ok_or(BeaconChainError::NoProposerForSlot(proposal_slot))?;
-
-            self.beacon_proposer_cache.lock().insert(
-                proposal_epoch,
-                decision_root,
-                proposers,
-                fork,
-            )?;
+        let shuffling_decision_root = cached_head
+            .snapshot
+            .beacon_state
+            .proposer_shuffling_decision_root_at_epoch(proposal_epoch, proposer_head, &self.spec)?;
+
+        let Some(proposer_index) = self.with_proposer_cache(
+            shuffling_decision_root,
+            proposal_epoch,
+            |proposers| proposers.get_slot::<T::EthSpec>(proposal_slot).map(|p| p.index as u64),
+            || {
+                if head_epoch + self.config.sync_tolerance_epochs < proposal_epoch {
+                    warn!(
+                        msg = "this is a non-critical issue that can happen on unhealthy nodes or \
+                               networks",
+                        %proposal_epoch,
+                        %head_epoch,
+                        "Skipping proposer preparation"
+                    );
 
-            // It's possible that the head changes whilst computing these duties. If so, abandon
-            // this routine since the change of head would have also spawned another instance of
-            // this routine.
-            //
-            // Exit now, after updating the cache.
-            if decision_root != shuffling_decision_root {
-                warn!("Head changed during proposer preparation");
-                return Ok(None);
+                    // Don't skip the head forward too many epochs. This avoids burdening an
+                    // unhealthy node.
+                    //
+                    // Although this node might miss out on preparing for a proposal, they should
+                    // still be able to propose. This will prioritise beacon chain health over
+                    // efficient packing of execution blocks.
+                    Err(Error::SkipProposerPreparation)
+                } else {
+                    let head = self.canonical_head.cached_head();
+                    Ok((
+                        head.head_state_root(),
+                        head.snapshot.beacon_state.clone(),
+                    ))
+                }
+            },
+        ).map_or_else(|e| {
+            match e {
+                Error::ProposerCacheIncorrectState { .. } => {
+                    warn!("Head changed during proposer preparation");
+                    Ok(None)
+                }
+                Error::SkipProposerPreparation => {
+                    // Warning logged for this above.
+                    Ok(None)
+                }
+                e => Err(e)
             }
-
-            proposer as u64
+        }, |value| Ok(Some(value)))? else {
+            return Ok(None);
         };
 
         // Get the `prev_randao` and parent block number.
@@ -4916,14 +4906,19 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
 
         // Only attempt a re-org if we have a proposer registered for the re-org slot.
         let proposing_at_re_org_slot = {
-            // The proposer shuffling has the same decision root as the next epoch attestation
-            // shuffling. We know our re-org block is not on the epoch boundary, so it has the
-            // same proposer shuffling as the head (but not necessarily the parent which may lie
-            // in the previous epoch).
-            let shuffling_decision_root = info
-                .head_node
-                .next_epoch_shuffling_id
-                .shuffling_decision_block;
+            // We know our re-org block is not on the epoch boundary, so it has the same proposer
+            // shuffling as the head (but not necessarily the parent which may lie in the previous
+            // epoch).
+            let shuffling_decision_root = if self
+                .spec
+                .fork_name_at_slot::<T::EthSpec>(re_org_block_slot)
+                .fulu_enabled()
+            {
+                info.head_node.current_epoch_shuffling_id
+            } else {
+                info.head_node.next_epoch_shuffling_id
+            }
+            .shuffling_decision_block;
             let proposer_index = self
                 .beacon_proposer_cache
                 .lock()
@@ -6558,6 +6553,70 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         }
     }
 
+    pub fn with_proposer_cache<V, E: From<BeaconChainError> + From<BeaconStateError>>(
+        &self,
+        shuffling_decision_block: Hash256,
+        proposal_epoch: Epoch,
+        accessor: impl Fn(&EpochBlockProposers) -> Result<V, BeaconChainError>,
+        state_provider: impl FnOnce() -> Result<(Hash256, BeaconState<T::EthSpec>), E>,
+    ) -> Result<V, E> {
+        let cache_entry = self
+            .beacon_proposer_cache
+            .lock()
+            .get_or_insert_key(proposal_epoch, shuffling_decision_block);
+
+        // If the cache entry is not initialised, run the code to initialise it inside a OnceCell.
+        // This prevents duplication of work across multiple threads.
+        //
+        // If it is already initialised, then `get_or_try_init` will return immediately without
+        // executing the initialisation code at all.
+        let epoch_block_proposers = cache_entry.get_or_try_init(|| {
+            debug!(
+                ?shuffling_decision_block,
+                %proposal_epoch,
+                "Proposer shuffling cache miss"
+            );
+
+            // Fetch the state on-demand if the required epoch was missing from the cache.
+            // If the caller wants to not compute the state they must return an error here and then
+            // catch it at the call site.
+            let (state_root, mut state) = state_provider()?;
+
+            // Ensure the state can compute proposer duties for `epoch`.
+            ensure_state_can_determine_proposers_for_epoch(
+                &mut state,
+                state_root,
+                proposal_epoch,
+                &self.spec,
+            )?;
+
+            // Sanity check the state.
+            let latest_block_root = state.get_latest_block_root(state_root);
+            let state_decision_block_root = state.proposer_shuffling_decision_root_at_epoch(
+                proposal_epoch,
+                latest_block_root,
+                &self.spec,
+            )?;
+            if state_decision_block_root != shuffling_decision_block {
+                return Err(Error::ProposerCacheIncorrectState {
+                    state_decision_block_root,
+                    requested_decision_block_root: shuffling_decision_block,
+                }
+                .into());
+            }
+
+            let proposers = state.get_beacon_proposer_indices(proposal_epoch, &self.spec)?;
+            Ok::<_, E>(EpochBlockProposers::new(
+                proposal_epoch,
+                state.fork(),
+                proposers,
+            ))
+        })?;
+
+        // Run the accessor function on the computed epoch proposers.
+        accessor(epoch_block_proposers).map_err(Into::into)
+    }
+
     /// Runs the `map_fn` with the committee cache for `shuffling_epoch` from the chain with head
     /// `head_block_root`. The `map_fn` will be supplied two values:
     ///

beacon_node/beacon_chain/src/beacon_proposer_cache.rs

@@ -12,9 +12,9 @@ use crate::{BeaconChain, BeaconChainError, BeaconChainTypes};
 use fork_choice::ExecutionStatus;
 use lru::LruCache;
 use once_cell::sync::OnceCell;
+use safe_arith::SafeArith;
 use smallvec::SmallVec;
 use state_processing::state_advance::partial_state_advance;
-use std::cmp::Ordering;
 use std::num::NonZeroUsize;
 use std::sync::Arc;
 use types::non_zero_usize::new_non_zero_usize;
@@ -51,6 +51,34 @@ pub struct EpochBlockProposers {
     pub(crate) proposers: SmallVec<[usize; TYPICAL_SLOTS_PER_EPOCH]>,
 }
 
+impl EpochBlockProposers {
+    pub fn new(epoch: Epoch, fork: Fork, proposers: Vec<usize>) -> Self {
+        Self {
+            epoch,
+            fork,
+            proposers: proposers.into(),
+        }
+    }
+
+    pub fn get_slot<E: EthSpec>(&self, slot: Slot) -> Result<Proposer, BeaconChainError> {
+        let epoch = slot.epoch(E::slots_per_epoch());
+        if epoch == self.epoch {
+            self.proposers
+                .get(slot.as_usize() % E::SlotsPerEpoch::to_usize())
+                .map(|&index| Proposer {
+                    index,
+                    fork: self.fork,
+                })
+                .ok_or(BeaconChainError::ProposerCacheOutOfBounds { slot, epoch })
+        } else {
+            Err(BeaconChainError::ProposerCacheWrongEpoch {
+                request_epoch: epoch,
+                cache_epoch: self.epoch,
+            })
+        }
+    }
+}
+
 /// A cache to store the proposers for some epoch.
 ///
 /// See the module-level documentation for more information.
@@ -76,23 +104,8 @@ impl BeaconProposerCache {
     ) -> Option<Proposer> {
         let epoch = slot.epoch(E::slots_per_epoch());
         let key = (epoch, shuffling_decision_block);
-        let cache_opt = self.cache.get(&key).and_then(|cell| cell.get());
-        if let Some(cache) = cache_opt {
-            // This `if` statement is likely unnecessary, but it feels like good practice.
-            if epoch == cache.epoch {
-                cache
-                    .proposers
-                    .get(slot.as_usize() % E::SlotsPerEpoch::to_usize())
-                    .map(|&index| Proposer {
-                        index,
-                        fork: cache.fork,
-                    })
-            } else {
-                None
-            }
-        } else {
-            None
-        }
+        let cache = self.cache.get(&key)?.get()?;
+        cache.get_slot::<E>(slot).ok()
     }
 
     /// As per `Self::get_slot`, but returns all proposers in all slots for the given `epoch`.
@@ -142,11 +155,7 @@ impl BeaconProposerCache {
     ) -> Result<(), BeaconStateError> {
         let key = (epoch, shuffling_decision_block);
         if !self.cache.contains(&key) {
-            let epoch_proposers = EpochBlockProposers {
-                epoch,
-                fork,
-                proposers: proposers.into(),
-            };
+            let epoch_proposers = EpochBlockProposers::new(epoch, fork, proposers);
             self.cache
                 .put(key, Arc::new(OnceCell::with_value(epoch_proposers)));
         }
@@ -178,44 +187,60 @@ pub fn compute_proposer_duties_from_head<T: BeaconChainTypes>(
         .ok_or(BeaconChainError::HeadMissingFromForkChoice(head_block_root))?;
 
     // Advance the state into the requested epoch.
-    ensure_state_is_in_epoch(&mut state, head_state_root, request_epoch, &chain.spec)?;
+    ensure_state_can_determine_proposers_for_epoch(
+        &mut state,
+        head_state_root,
+        request_epoch,
+        &chain.spec,
+    )?;
 
     let indices = state
         .get_beacon_proposer_indices(request_epoch, &chain.spec)
         .map_err(BeaconChainError::from)?;
 
     let dependent_root = state
         // The only block which decides its own shuffling is the genesis block.
-        .proposer_shuffling_decision_root(chain.genesis_block_root)
+        .proposer_shuffling_decision_root(chain.genesis_block_root, &chain.spec)
         .map_err(BeaconChainError::from)?;
 
     Ok((indices, dependent_root, execution_status, state.fork()))
 }
 
-/// If required, advance `state` to `target_epoch`.
+/// If required, advance `state` to the epoch required to determine proposer indices in `target_epoch`.
 ///
 /// ## Details
 ///
 /// - Returns an error if `state.current_epoch() > target_epoch`.
 /// - No-op if `state.current_epoch() == target_epoch`.
 /// - It must be the case that `state.canonical_root() == state_root`, but this function will not
 ///   check that.
-pub fn ensure_state_is_in_epoch<E: EthSpec>(
+pub fn ensure_state_can_determine_proposers_for_epoch<E: EthSpec>(
     state: &mut BeaconState<E>,
     state_root: Hash256,
     target_epoch: Epoch,
     spec: &ChainSpec,
 ) -> Result<(), BeaconChainError> {
-    match state.current_epoch().cmp(&target_epoch) {
-        // Protects against an inconsistent slot clock.
-        Ordering::Greater => Err(BeaconStateError::SlotOutOfBounds.into()),
-        // The state needs to be advanced.
-        Ordering::Less => {
-            let target_slot = target_epoch.start_slot(E::slots_per_epoch());
-            partial_state_advance(state, Some(state_root), target_slot, spec)
-                .map_err(BeaconChainError::from)
-        }
-        // The state is suitable, nothing to do.
-        Ordering::Equal => Ok(()),
+    // The decision slot is the end of an epoch, so we add 1 to reach the first slot of the epoch
+    // at which the shuffling is determined.
+    let minimum_slot = spec
+        .proposer_shuffling_decision_slot::<E>(target_epoch)
+        .safe_add(1)?;
+    let minimum_epoch = minimum_slot.epoch(E::slots_per_epoch());
+
+    // Before and after Fulu, the oldest epoch reachable from a state at epoch N is epoch N itself,
+    // i.e. we can never "look back".
+    let maximum_epoch = target_epoch;
+
+    if state.current_epoch() > maximum_epoch {
+        Err(BeaconStateError::SlotOutOfBounds.into())
+    } else if state.current_epoch() >= minimum_epoch {
+        // Fulu allows us to access shufflings in multiple epochs (thanks to lookahead).
+        // Pre-Fulu we expect `minimum_epoch == maximum_epoch`, and this branch covers that case.
+        Ok(())
+    } else {
+        // State's current epoch is less than the minimum epoch.
+        // Advance the state up to the minimum epoch.
+        partial_state_advance(state, Some(state_root), minimum_slot, spec)
+            .map_err(BeaconChainError::from)
     }
 }