Cache indexed attestations

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -140,10 +140,6 @@ const MAX_PER_SLOT_FORK_CHOICE_DISTANCE: u64 = 256;
 pub const INVALID_JUSTIFIED_PAYLOAD_SHUTDOWN_REASON: &str =
     "Justified block has an invalid execution payload.";
 
-// FIXME(sproul): decide whether to keep this
-// Interval before the attestation deadline during which to consider blocks "borderline" late.
-// const BORDERLINE_LATE_BLOCK_TOLERANCE: Duration = Duration::from_millis(50);
-
 pub const INVALID_FINALIZED_MERGE_TRANSITION_BLOCK_SHUTDOWN_REASON: &str =
     "Finalized merge transition block is invalid.";
 
@@ -3369,10 +3365,12 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         // This will be a lot slower but guards against bugs in block production and can be
         // quickly rolled out without a release.
         if self.config.paranoid_block_proposal {
+            let mut tmp_ctxt = ConsensusContext::new(state.slot());
             attestations.retain(|att| {
                 verify_attestation_for_block_inclusion(
                     &state,
                     att,
+                    &mut tmp_ctxt,
                     VerifySignatures::True,
                     &self.spec,
                 )

beacon_node/beacon_chain/src/block_verification.rs

@@ -547,8 +547,22 @@ pub fn signature_verify_chain_segment<T: BeaconChainTypes>(
     let pubkey_cache = get_validator_pubkey_cache(chain)?;
     let mut signature_verifier = get_signature_verifier::<T>(&state, &pubkey_cache, &chain.spec);
 
+    let mut signature_verified_blocks = Vec::with_capacity(chain_segment.len());
+
     for (block_root, block) in &chain_segment {
-        signature_verifier.include_all_signatures(block, Some(*block_root), None)?;
+        let mut consensus_context =
+            ConsensusContext::new(block.slot()).set_current_block_root(*block_root);
+
+        signature_verifier.include_all_signatures(&block, &mut consensus_context)?;
+
+        // Save the block and its consensus context. The context will have had its proposer index
+        // and attesting indices filled in, which can be used to accelerate later block processing.
+        signature_verified_blocks.push(SignatureVerifiedBlock {
+            block: block.clone(),
+            block_root: *block_root,
+            parent: None,
+            consensus_context,
+        });
     }
 
     if signature_verifier.verify().is_err() {
@@ -557,22 +571,6 @@ pub fn signature_verify_chain_segment<T: BeaconChainTypes>(
 
     drop(pubkey_cache);
 
-    let mut signature_verified_blocks = chain_segment
-        .into_iter()
-        .map(|(block_root, block)| {
-            // Proposer index has already been verified above during signature verification.
-            let consensus_context = ConsensusContext::new(block.slot())
-                .set_current_block_root(block_root)
-                .set_proposer_index(block.message().proposer_index());
-            SignatureVerifiedBlock {
-                block,
-                block_root,
-                parent: None,
-                consensus_context,
-            }
-        })
-        .collect::<Vec<_>>();
-
     if let Some(signature_verified_block) = signature_verified_blocks.first_mut() {
         signature_verified_block.parent = Some(parent);
     }
@@ -941,13 +939,14 @@ impl<T: BeaconChainTypes> SignatureVerifiedBlock<T> {
         let mut signature_verifier =
             get_signature_verifier::<T>(&state, &pubkey_cache, &chain.spec);
 
-        signature_verifier.include_all_signatures(&block, Some(block_root), None)?;
+        let mut consensus_context =
+            ConsensusContext::new(block.slot()).set_current_block_root(block_root);
+
+        signature_verifier.include_all_signatures(&block, &mut consensus_context)?;
 
         if signature_verifier.verify().is_ok() {
             Ok(Self {
-                consensus_context: ConsensusContext::new(block.slot())
+                consensus_context,
-                    .set_current_block_root(block_root)
-                    .set_proposer_index(block.message().proposer_index()),
                 block,
                 block_root,
                 parent: Some(parent),
@@ -993,16 +992,16 @@ impl<T: BeaconChainTypes> SignatureVerifiedBlock<T> {
 
         // Gossip verification has already checked the proposer index. Use it to check the RANDAO
         // signature.
-        let verified_proposer_index = Some(block.message().proposer_index());
+        let mut consensus_context = from.consensus_context;
         signature_verifier
-            .include_all_signatures_except_proposal(&block, verified_proposer_index)?;
+            .include_all_signatures_except_proposal(&block, &mut consensus_context)?;
 
         if signature_verifier.verify().is_ok() {
             Ok(Self {
                 block,
                 block_root: from.block_root,
                 parent: Some(parent),
-                consensus_context: from.consensus_context,
+                consensus_context,
             })
         } else {
             Err(BlockError::InvalidSignature)

consensus/ssz_types/src/bitfield.rs

@@ -22,7 +22,7 @@ pub trait BitfieldBehaviour: Clone {}
 /// A marker struct used to declare SSZ `Variable` behaviour on a `Bitfield`.
 ///
 /// See the [`Bitfield`](struct.Bitfield.html) docs for usage.
-#[derive(Clone, PartialEq, Debug)]
+#[derive(Clone, PartialEq, Eq, Debug)]
 pub struct Variable<N> {
     _phantom: PhantomData<N>,
 }
@@ -30,7 +30,7 @@ pub struct Variable<N> {
 /// A marker struct used to declare SSZ `Fixed` behaviour on a `Bitfield`.
 ///
 /// See the [`Bitfield`](struct.Bitfield.html) docs for usage.
-#[derive(Clone, PartialEq, Debug)]
+#[derive(Clone, PartialEq, Eq, Debug)]
 pub struct Fixed<N> {
     _phantom: PhantomData<N>,
 }
@@ -96,7 +96,7 @@ pub type BitVector<N> = Bitfield<Fixed<N>>;
 /// byte (by `Vec` index) stores the lowest bit-indices and the right-most bit stores the lowest
 /// bit-index. E.g., `smallvec![0b0000_0001, 0b0000_0010]` has bits `0, 9` set.
 #[derive(Clone, Debug, Derivative)]
-#[derivative(PartialEq, Hash(bound = ""))]
+#[derivative(PartialEq, Eq, Hash(bound = ""))]
 pub struct Bitfield<T> {
     bytes: SmallVec<[u8; SMALLVEC_LEN]>,
     len: usize,

consensus/state_processing/src/consensus_context.rs

@@ -1,10 +1,13 @@
+use crate::common::get_indexed_attestation;
+use crate::per_block_processing::errors::{AttestationInvalid, BlockOperationError};
 use crate::{EpochCache, EpochCacheError};
 use std::borrow::Cow;
+use std::collections::{hash_map::Entry, HashMap};
 use std::marker::PhantomData;
 use tree_hash::TreeHash;
 use types::{
-    BeaconState, BeaconStateError, ChainSpec, EthSpec, ExecPayload, Hash256, SignedBeaconBlock,
+    Attestation, AttestationData, BeaconState, BeaconStateError, BitList, ChainSpec, EthSpec,
-    Slot,
+    ExecPayload, Hash256, IndexedAttestation, SignedBeaconBlock, Slot,
 };
 
 #[derive(Debug, Clone)]
@@ -17,6 +20,9 @@ pub struct ConsensusContext<T: EthSpec> {
     current_block_root: Option<Hash256>,
     /// Epoch cache of values that are useful for block processing that are static over an epoch.
     epoch_cache: Option<EpochCache>,
+    /// Cache of indexed attestations constructed during block processing.
+    indexed_attestations:
+        HashMap<(AttestationData, BitList<T::MaxValidatorsPerCommittee>), IndexedAttestation<T>>,
     _phantom: PhantomData<T>,
 }
 
@@ -46,6 +52,7 @@ impl<T: EthSpec> ConsensusContext<T> {
             proposer_index: None,
             current_block_root: None,
             epoch_cache: None,
+            indexed_attestations: HashMap::new(),
             _phantom: PhantomData,
         }
     }
@@ -107,9 +114,9 @@ impl<T: EthSpec> ConsensusContext<T> {
         self
     }
 
-    pub fn get_base_reward<E: EthSpec>(
+    pub fn get_base_reward(
         &mut self,
-        state: &BeaconState<E>,
+        state: &BeaconState<T>,
         validator_index: usize,
         spec: &ChainSpec,
     ) -> Result<u64, ContextError> {
@@ -126,4 +133,30 @@ impl<T: EthSpec> ConsensusContext<T> {
 
         Ok(epoch_cache.get_base_reward(validator_index)?)
     }
+
+    pub fn get_indexed_attestation(
+        &mut self,
+        state: &BeaconState<T>,
+        attestation: &Attestation<T>,
+    ) -> Result<&IndexedAttestation<T>, BlockOperationError<AttestationInvalid>> {
+        let key = (
+            attestation.data.clone(),
+            attestation.aggregation_bits.clone(),
+        );
+
+        match self.indexed_attestations.entry(key) {
+            Entry::Occupied(occupied) => Ok(occupied.into_mut()),
+            Entry::Vacant(vacant) => {
+                let committee =
+                    state.get_beacon_committee(attestation.data.slot, attestation.data.index)?;
+                let indexed_attestation =
+                    get_indexed_attestation(committee.committee, attestation)?;
+                Ok(vacant.insert(indexed_attestation))
+            }
+        }
+    }
+
+    pub fn num_cached_indexed_attestations(&self) -> usize {
+        self.indexed_attestations.len()
+    }
 }

consensus/state_processing/src/per_block_processing.rs

@@ -111,16 +111,13 @@ pub fn per_block_processing<T: EthSpec, Payload: ExecPayload<T>>(
     let verify_signatures = match block_signature_strategy {
         BlockSignatureStrategy::VerifyBulk => {
             // Verify all signatures in the block at once.
-            let block_root = Some(ctxt.get_current_block_root(signed_block)?);
-            let proposer_index = Some(ctxt.get_proposer_index(state, spec)?);
             block_verify!(
                 BlockSignatureVerifier::verify_entire_block(
                     state,
                     |i| get_pubkey_from_state(state, i),
                     |pk_bytes| pk_bytes.decompress().ok().map(Cow::Owned),
                     signed_block,
-                    block_root,
+                    ctxt,
-                    proposer_index,
                     spec
                 )
                 .is_ok(),

consensus/state_processing/src/per_block_processing/block_signature_verifier.rs

@@ -1,14 +1,13 @@
 #![allow(clippy::integer_arithmetic)]
 
 use super::signature_sets::{Error as SignatureSetError, *};
-use crate::common::get_indexed_attestation;
 use crate::per_block_processing::errors::{AttestationInvalid, BlockOperationError};
+use crate::{ConsensusContext, ContextError};
 use bls::{verify_signature_sets, PublicKey, PublicKeyBytes, SignatureSet};
 use rayon::prelude::*;
 use std::borrow::Cow;
 use types::{
-    BeaconState, BeaconStateError, ChainSpec, EthSpec, ExecPayload, Hash256, IndexedAttestation,
+    BeaconState, BeaconStateError, ChainSpec, EthSpec, ExecPayload, Hash256, SignedBeaconBlock,
-    SignedBeaconBlock,
 };
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -28,6 +27,8 @@ pub enum Error {
     IncorrectBlockProposer { block: u64, local_shuffling: u64 },
     /// Failed to load a signature set. The block may be invalid or we failed to process it.
     SignatureSetError(SignatureSetError),
+    /// Error related to the consensus context, likely the proposer index or block root calc.
+    ContextError(ContextError),
 }
 
 impl From<BeaconStateError> for Error {
@@ -36,6 +37,12 @@ impl From<BeaconStateError> for Error {
     }
 }
 
+impl From<ContextError> for Error {
+    fn from(e: ContextError) -> Error {
+        Error::ContextError(e)
+    }
+}
+
 impl From<SignatureSetError> for Error {
     fn from(e: SignatureSetError) -> Error {
         match e {
@@ -122,12 +129,11 @@ where
         get_pubkey: F,
         decompressor: D,
         block: &'a SignedBeaconBlock<T, Payload>,
-        block_root: Option<Hash256>,
+        ctxt: &mut ConsensusContext<T>,
-        verified_proposer_index: Option<u64>,
         spec: &'a ChainSpec,
     ) -> Result<()> {
         let mut verifier = Self::new(state, get_pubkey, decompressor, spec);
-        verifier.include_all_signatures(block, block_root, verified_proposer_index)?;
+        verifier.include_all_signatures(block, ctxt)?;
         verifier.verify()
     }
 
@@ -135,11 +141,13 @@ where
     pub fn include_all_signatures<Payload: ExecPayload<T>>(
         &mut self,
         block: &'a SignedBeaconBlock<T, Payload>,
-        block_root: Option<Hash256>,
+        ctxt: &mut ConsensusContext<T>,
-        verified_proposer_index: Option<u64>,
     ) -> Result<()> {
+        let block_root = Some(ctxt.get_current_block_root(block)?);
+        let verified_proposer_index = Some(ctxt.get_proposer_index(self.state, self.spec)?);
+
         self.include_block_proposal(block, block_root, verified_proposer_index)?;
-        self.include_all_signatures_except_proposal(block, verified_proposer_index)?;
+        self.include_all_signatures_except_proposal(block, ctxt)?;
 
         Ok(())
     }
@@ -149,12 +157,13 @@ where
     pub fn include_all_signatures_except_proposal<Payload: ExecPayload<T>>(
         &mut self,
         block: &'a SignedBeaconBlock<T, Payload>,
-        verified_proposer_index: Option<u64>,
+        ctxt: &mut ConsensusContext<T>,
     ) -> Result<()> {
+        let verified_proposer_index = Some(ctxt.get_proposer_index(self.state, self.spec)?);
         self.include_randao_reveal(block, verified_proposer_index)?;
         self.include_proposer_slashings(block)?;
         self.include_attester_slashings(block)?;
-        self.include_attestations(block)?;
+        self.include_attestations(block, ctxt)?;
         // Deposits are not included because they can legally have invalid signatures.
         self.include_exits(block)?;
         self.include_sync_aggregate(block)?;
@@ -260,7 +269,8 @@ where
     pub fn include_attestations<Payload: ExecPayload<T>>(
         &mut self,
         block: &'a SignedBeaconBlock<T, Payload>,
-    ) -> Result<Vec<IndexedAttestation<T>>> {
+        ctxt: &mut ConsensusContext<T>,
+    ) -> Result<()> {
         self.sets
             .sets
             .reserve(block.message().body().attestations().len());
@@ -270,28 +280,18 @@ where
             .body()
             .attestations()
             .iter()
-            .try_fold(
+            .try_for_each(|attestation| {
-                Vec::with_capacity(block.message().body().attestations().len()),
+                let indexed_attestation = ctxt.get_indexed_attestation(&self.state, attestation)?;
-                |mut vec, attestation| {
-                    let committee = self
-                        .state
-                        .get_beacon_committee(attestation.data.slot, attestation.data.index)?;
-                    let indexed_attestation =
-                        get_indexed_attestation(committee.committee, attestation)?;
 
-                    self.sets.push(indexed_attestation_signature_set(
+                self.sets.push(indexed_attestation_signature_set(
-                        self.state,
+                    self.state,
-                        self.get_pubkey.clone(),
+                    self.get_pubkey.clone(),
-                        &attestation.signature,
+                    &attestation.signature,
-                        &indexed_attestation,
+                    indexed_attestation,
-                        self.spec,
+                    self.spec,
-                    )?);
+                )?);
-
+                Ok(())
-                    vec.push(indexed_attestation);
+            })
-
-                    Ok(vec)
-                },
-            )
             .map_err(Error::into)
     }
 

consensus/state_processing/src/per_block_processing/process_operations.rs

@@ -57,8 +57,14 @@ pub mod base {
 
         // Verify and apply each attestation.
         for (i, attestation) in attestations.iter().enumerate() {
-            verify_attestation_for_block_inclusion(state, attestation, verify_signatures, spec)
+            verify_attestation_for_block_inclusion(
-                .map_err(|e| e.into_with_index(i))?;
+                state,
+                attestation,
+                ctxt,
+                verify_signatures,
+                spec,
+            )
+            .map_err(|e| e.into_with_index(i))?;
 
             let pending_attestation = PendingAttestation {
                 aggregation_bits: attestation.aggregation_bits.clone(),
@@ -115,9 +121,16 @@ pub mod altair {
 
         let proposer_index = ctxt.get_proposer_index(state, spec)?;
 
-        let indexed_attestation =
+        let attesting_indices = verify_attestation_for_block_inclusion(
-            verify_attestation_for_block_inclusion(state, attestation, verify_signatures, spec)
+            state,
-                .map_err(|e| e.into_with_index(att_index))?;
+            attestation,
+            ctxt,
+            verify_signatures,
+            spec,
+        )
+        .map_err(|e| e.into_with_index(att_index))?
+        .attesting_indices
+        .clone();
 
         // Matching roots, participation flag indices
         let data = &attestation.data;
@@ -127,7 +140,7 @@ pub mod altair {
 
         // Update epoch participation flags.
         let mut proposer_reward_numerator = 0;
-        for index in &indexed_attestation.attesting_indices {
+        for index in &attesting_indices {
             let index = *index as usize;
 
             for (flag_index, &weight) in PARTICIPATION_FLAG_WEIGHTS.iter().enumerate() {

consensus/state_processing/src/per_block_processing/verify_attestation.rs

@@ -1,7 +1,7 @@
 use super::errors::{AttestationInvalid as Invalid, BlockOperationError};
 use super::VerifySignatures;
-use crate::common::get_indexed_attestation;
 use crate::per_block_processing::is_valid_indexed_attestation;
+use crate::ConsensusContext;
 use safe_arith::SafeArith;
 use types::*;
 
@@ -15,12 +15,13 @@ fn error(reason: Invalid) -> BlockOperationError<Invalid> {
 /// to `state`. Otherwise, returns a descriptive `Err`.
 ///
 /// Optionally verifies the aggregate signature, depending on `verify_signatures`.
-pub fn verify_attestation_for_block_inclusion<T: EthSpec>(
+pub fn verify_attestation_for_block_inclusion<'ctxt, T: EthSpec>(
     state: &BeaconState<T>,
     attestation: &Attestation<T>,
+    ctxt: &'ctxt mut ConsensusContext<T>,
     verify_signatures: VerifySignatures,
     spec: &ChainSpec,
-) -> Result<IndexedAttestation<T>> {
+) -> Result<&'ctxt IndexedAttestation<T>> {
     let data = &attestation.data;
 
     verify!(
@@ -39,7 +40,7 @@ pub fn verify_attestation_for_block_inclusion<T: EthSpec>(
         }
     );
 
-    verify_attestation_for_state(state, attestation, verify_signatures, spec)
+    verify_attestation_for_state(state, attestation, ctxt, verify_signatures, spec)
 }
 
 /// Returns `Ok(())` if `attestation` is a valid attestation to the chain that precedes the given
@@ -49,12 +50,13 @@ pub fn verify_attestation_for_block_inclusion<T: EthSpec>(
 /// prior blocks in `state`.
 ///
 /// Spec v0.12.1
-pub fn verify_attestation_for_state<T: EthSpec>(
+pub fn verify_attestation_for_state<'ctxt, T: EthSpec>(
     state: &BeaconState<T>,
     attestation: &Attestation<T>,
+    ctxt: &'ctxt mut ConsensusContext<T>,
     verify_signatures: VerifySignatures,
     spec: &ChainSpec,
-) -> Result<IndexedAttestation<T>> {
+) -> Result<&'ctxt IndexedAttestation<T>> {
     let data = &attestation.data;
 
     verify!(
@@ -66,9 +68,8 @@ pub fn verify_attestation_for_state<T: EthSpec>(
     verify_casper_ffg_vote(attestation, state)?;
 
     // Check signature and bitfields
-    let committee = state.get_beacon_committee(attestation.data.slot, attestation.data.index)?;
+    let indexed_attestation = ctxt.get_indexed_attestation(state, attestation)?;
-    let indexed_attestation = get_indexed_attestation(committee.committee, attestation)?;
+    is_valid_indexed_attestation(state, indexed_attestation, verify_signatures, spec)?;
-    is_valid_indexed_attestation(state, &indexed_attestation, verify_signatures, spec)?;
 
     Ok(indexed_attestation)
 }

lcli/src/transition_blocks.rs

@@ -347,6 +347,23 @@ fn do_transition<T: EthSpec>(
         .map_err(|e| format!("Unable to build caches: {:?}", e))?;
     debug!("Build all caches (again): {:?}", t.elapsed());
 
+    let mut ctxt = if let Some(ctxt) = saved_ctxt {
+        ctxt.clone()
+    } else {
+        let mut ctxt = ConsensusContext::new(pre_state.slot())
+            .set_current_block_root(block_root)
+            .set_proposer_index(block.message().proposer_index());
+
+        if config.exclude_cache_builds {
+            ctxt = ctxt.set_epoch_cache(
+                EpochCache::new(&pre_state, spec)
+                    .map_err(|e| format!("unable to build epoch cache: {e:?}"))?,
+            );
+            *saved_ctxt = Some(ctxt.clone());
+        }
+        ctxt
+    };
+
     if !config.no_signature_verification {
         let get_pubkey = move |validator_index| {
             validator_pubkey_cache
@@ -367,31 +384,19 @@ fn do_transition<T: EthSpec>(
             get_pubkey,
             decompressor,
             &block,
-            Some(block_root),
+            &mut ctxt,
-            Some(block.message().proposer_index()),
             spec,
         )
         .map_err(|e| format!("Invalid block signature: {:?}", e))?;
         debug!("Batch verify block signatures: {:?}", t.elapsed());
+
+        // Signature verification should prime the indexed attestation cache.
+        assert_eq!(
+            ctxt.num_cached_indexed_attestations(),
+            block.message().body().attestations().len()
+        );
     }
 
-    let mut ctxt = if let Some(ctxt) = saved_ctxt {
-        ctxt.clone()
-    } else {
-        let mut ctxt = ConsensusContext::new(pre_state.slot())
-            .set_current_block_root(block_root)
-            .set_proposer_index(block.message().proposer_index());
-
-        if config.exclude_cache_builds {
-            ctxt = ctxt.set_epoch_cache(
-                EpochCache::new(&pre_state, spec)
-                    .map_err(|e| format!("unable to build epoch cache: {e:?}"))?,
-            );
-            *saved_ctxt = Some(ctxt.clone());
-        }
-        ctxt
-    };
-
     let t = Instant::now();
 
     per_block_processing(