Make block production async (make lint passing)

michaelsproul · Jun 28, 2022 · 59351d2 · 59351d2
1 parent 689f3e6
commit 59351d2
Show file tree

Hide file tree

Showing 8 changed files with 362 additions and 181 deletions.
diff --git a/beacon_node/beacon_chain/src/beacon_chain.rs b/beacon_node/beacon_chain/src/beacon_chain.rs
diff --git a/beacon_node/beacon_chain/src/errors.rs b/beacon_node/beacon_chain/src/errors.rs
@@ -248,6 +248,9 @@ pub enum BlockProductionError {
     BlockTooLarge(usize),
     ForkChoiceError(BeaconChainError),
     ShuttingDown,
+    MissingSyncAggregate,
+    MissingExecutionPayload,
+    TokioJoin(tokio::task::JoinError),
 }
 
 easy_from_to!(BlockProcessingError, BlockProductionError);

diff --git a/beacon_node/beacon_chain/src/execution_payload.rs b/beacon_node/beacon_chain/src/execution_payload.rs
@@ -24,8 +24,8 @@ use std::sync::Arc;
 use tokio::task::JoinHandle;
 use types::*;
 
-type PreparePayloadResult<Payload> = Result<Payload, BlockProductionError>;
-type PreparePayloadHandle<Payload> = JoinHandle<Option<PreparePayloadResult<Payload>>>;
+pub type PreparePayloadResult<Payload> = Result<Payload, BlockProductionError>;
+pub type PreparePayloadHandle<Payload> = JoinHandle<Option<PreparePayloadResult<Payload>>>;
 
 /// Used to await the result of executing payload with a remote EE.
 pub struct PayloadNotifier<T: BeaconChainTypes> {
@@ -342,6 +342,7 @@ pub fn get_execution_payload<
 /// Equivalent to the `prepare_execution_payload` function in the Validator Guide:
 ///
 /// https://github.com/ethereum/consensus-specs/blob/v1.1.5/specs/merge/validator.md#block-proposal
+#[allow(clippy::too_many_arguments)]
 pub async fn prepare_execution_payload<T, Payload>(
     chain: &BeaconChain<T>,
     current_epoch: Epoch,

diff --git a/beacon_node/beacon_chain/src/test_utils.rs b/beacon_node/beacon_chain/src/test_utils.rs
@@ -580,7 +580,7 @@ where
         state.get_block_root(slot).unwrap() == state.get_block_root(slot - 1).unwrap()
     }
 
-    pub fn make_block(
+    pub async fn make_block(
         &self,
         mut state: BeaconState<E>,
         slot: Slot,
@@ -614,6 +614,7 @@ where
                 Some(graffiti),
                 ProduceBlockVerification::VerifyRandao,
             )
+            .await
             .unwrap();
 
         let signed_block = block.sign(
@@ -628,7 +629,7 @@ where
 
     /// Useful for the `per_block_processing` tests. Creates a block, and returns the state after
     /// caches are built but before the generated block is processed.
-    pub fn make_block_return_pre_state(
+    pub async fn make_block_return_pre_state(
         &self,
         mut state: BeaconState<E>,
         slot: Slot,
@@ -664,6 +665,7 @@ where
                 Some(graffiti),
                 ProduceBlockVerification::VerifyRandao,
             )
+            .await
             .unwrap();
 
         let signed_block = block.sign(
@@ -1250,7 +1252,7 @@ where
     /// Create a new block, apply `block_modifier` to it, sign it and return it.
     ///
     /// The state returned is a pre-block state at the same slot as the produced block.
-    pub fn make_block_with_modifier(
+    pub async fn make_block_with_modifier(
         &self,
         state: BeaconState<E>,
         slot: Slot,
@@ -1259,7 +1261,7 @@ where
         assert_ne!(slot, 0, "can't produce a block at slot 0");
         assert!(slot >= state.slot());
 
-        let (block, state) = self.make_block_return_pre_state(state, slot);
+        let (block, state) = self.make_block_return_pre_state(state, slot).await;
         let (mut block, _) = block.deconstruct();
 
         block_modifier(&mut block);
@@ -1426,7 +1428,7 @@ where
         state: BeaconState<E>,
     ) -> Result<(SignedBeaconBlockHash, SignedBeaconBlock<E>, BeaconState<E>), BlockError<E>> {
         self.set_current_slot(slot);
-        let (block, new_state) = self.make_block(state, slot);
+        let (block, new_state) = self.make_block(state, slot).await;
         let block_hash = self.process_block(slot, block.clone()).await?;
         Ok((block_hash, block, new_state))
     }
@@ -1616,13 +1618,13 @@ where
     /// Deprecated: Use make_block() instead
     ///
     /// Returns a newly created block, signed by the proposer for the given slot.
-    pub fn build_block(
+    pub async fn build_block(
         &self,
         state: BeaconState<E>,
         slot: Slot,
         _block_strategy: BlockStrategy,
     ) -> (SignedBeaconBlock<E>, BeaconState<E>) {
-        self.make_block(state, slot)
+        self.make_block(state, slot).await
     }
 
     /// Uses `Self::extend_chain` to build the chain out to the `target_slot`.

diff --git a/beacon_node/http_api/src/lib.rs b/beacon_node/http_api/src/lib.rs
@@ -1942,48 +1942,49 @@ pub fn serve<T: BeaconChainTypes>(
             |endpoint_version: EndpointVersion,
              slot: Slot,
              query: api_types::ValidatorBlocksQuery,
-             chain: Arc<BeaconChain<T>>| {
-                blocking_json_task(move || {
-                    let randao_reveal = query.randao_reveal.as_ref().map_or_else(
-                        || {
-                            if query.verify_randao {
-                                Err(warp_utils::reject::custom_bad_request(
-                                    "randao_reveal is mandatory unless verify_randao=false".into(),
-                                ))
-                            } else {
-                                Ok(Signature::empty())
-                            }
-                        },
-                        |sig_bytes| {
-                            sig_bytes.try_into().map_err(|e| {
-                                warp_utils::reject::custom_bad_request(format!(
-                                    "randao reveal is not a valid BLS signature: {:?}",
-                                    e
-                                ))
-                            })
-                        },
-                    )?;
+             chain: Arc<BeaconChain<T>>| async move {
+                let randao_reveal = query.randao_reveal.as_ref().map_or_else(
+                    || {
+                        if query.verify_randao {
+                            Err(warp_utils::reject::custom_bad_request(
+                                "randao_reveal is mandatory unless verify_randao=false".into(),
+                            ))
+                        } else {
+                            Ok(Signature::empty())
+                        }
+                    },
+                    |sig_bytes| {
+                        sig_bytes.try_into().map_err(|e| {
+                            warp_utils::reject::custom_bad_request(format!(
+                                "randao reveal is not a valid BLS signature: {:?}",
+                                e
+                            ))
+                        })
+                    },
+                )?;
 
-                    let randao_verification = if query.verify_randao {
-                        ProduceBlockVerification::VerifyRandao
-                    } else {
-                        ProduceBlockVerification::NoVerification
-                    };
+                let randao_verification = if query.verify_randao {
+                    ProduceBlockVerification::VerifyRandao
+                } else {
+                    ProduceBlockVerification::NoVerification
+                };
 
-                    let (block, _) = chain
-                        .produce_block_with_verification::<FullPayload<T::EthSpec>>(
-                            randao_reveal,
-                            slot,
-                            query.graffiti.map(Into::into),
-                            randao_verification,
-                        )
-                        .map_err(warp_utils::reject::block_production_error)?;
-                    let fork_name = block
-                        .to_ref()
-                        .fork_name(&chain.spec)
-                        .map_err(inconsistent_fork_rejection)?;
-                    fork_versioned_response(endpoint_version, fork_name, block)
-                })
+                let (block, _) = chain
+                    .produce_block_with_verification::<FullPayload<T::EthSpec>>(
+                        randao_reveal,
+                        slot,
+                        query.graffiti.map(Into::into),
+                        randao_verification,
+                    )
+                    .await
+                    .map_err(warp_utils::reject::block_production_error)?;
+                let fork_name = block
+                    .to_ref()
+                    .fork_name(&chain.spec)
+                    .map_err(inconsistent_fork_rejection)?;
+
+                fork_versioned_response(endpoint_version, fork_name, block)
+                    .map(|response| warp::reply::json(&response))
             },
         );
 
@@ -2004,48 +2005,48 @@ pub fn serve<T: BeaconChainTypes>(
             |endpoint_version: EndpointVersion,
              slot: Slot,
              query: api_types::ValidatorBlocksQuery,
-             chain: Arc<BeaconChain<T>>| {
-                blocking_json_task(move || {
-                    let randao_reveal = query.randao_reveal.as_ref().map_or_else(
-                        || {
-                            if query.verify_randao {
-                                Err(warp_utils::reject::custom_bad_request(
-                                    "randao_reveal is mandatory unless verify_randao=false".into(),
-                                ))
-                            } else {
-                                Ok(Signature::empty())
-                            }
-                        },
-                        |sig_bytes| {
-                            sig_bytes.try_into().map_err(|e| {
-                                warp_utils::reject::custom_bad_request(format!(
-                                    "randao reveal is not a valid BLS signature: {:?}",
-                                    e
-                                ))
-                            })
-                        },
-                    )?;
+             chain: Arc<BeaconChain<T>>| async move {
+                let randao_reveal = query.randao_reveal.as_ref().map_or_else(
+                    || {
+                        if query.verify_randao {
+                            Err(warp_utils::reject::custom_bad_request(
+                                "randao_reveal is mandatory unless verify_randao=false".into(),
+                            ))
+                        } else {
+                            Ok(Signature::empty())
+                        }
+                    },
+                    |sig_bytes| {
+                        sig_bytes.try_into().map_err(|e| {
+                            warp_utils::reject::custom_bad_request(format!(
+                                "randao reveal is not a valid BLS signature: {:?}",
+                                e
+                            ))
+                        })
+                    },
+                )?;
 
-                    let randao_verification = if query.verify_randao {
-                        ProduceBlockVerification::VerifyRandao
-                    } else {
-                        ProduceBlockVerification::NoVerification
-                    };
+                let randao_verification = if query.verify_randao {
+                    ProduceBlockVerification::VerifyRandao
+                } else {
+                    ProduceBlockVerification::NoVerification
+                };
 
-                    let (block, _) = chain
-                        .produce_block_with_verification::<BlindedPayload<T::EthSpec>>(
-                            randao_reveal,
-                            slot,
-                            query.graffiti.map(Into::into),
-                            randao_verification,
-                        )
-                        .map_err(warp_utils::reject::block_production_error)?;
-                    let fork_name = block
-                        .to_ref()
-                        .fork_name(&chain.spec)
-                        .map_err(inconsistent_fork_rejection)?;
-                    fork_versioned_response(endpoint_version, fork_name, block)
-                })
+                let (block, _) = chain
+                    .produce_block_with_verification::<BlindedPayload<T::EthSpec>>(
+                        randao_reveal,
+                        slot,
+                        query.graffiti.map(Into::into),
+                        randao_verification,
+                    )
+                    .await
+                    .map_err(warp_utils::reject::block_production_error)?;
+                let fork_name = block
+                    .to_ref()
+                    .fork_name(&chain.spec)
+                    .map_err(inconsistent_fork_rejection)?;
+                fork_versioned_response(endpoint_version, fork_name, block)
+                    .map(|response| warp::reply::json(&response))
             },
         );
 

diff --git a/consensus/state_processing/src/per_block_processing/tests.rs b/consensus/state_processing/src/per_block_processing/tests.rs
@@ -63,7 +63,9 @@ async fn valid_block_ok() {
     let state = harness.get_current_state();
 
     let slot = state.slot();
-    let (block, mut state) = harness.make_block_return_pre_state(state, slot + Slot::new(1));
+    let (block, mut state) = harness
+        .make_block_return_pre_state(state, slot + Slot::new(1))
+        .await;
 
     let result = per_block_processing(
         &mut state,
@@ -85,7 +87,7 @@ async fn invalid_block_header_state_slot() {
     let state = harness.get_current_state();
     let slot = state.slot() + Slot::new(1);
 
-    let (signed_block, mut state) = harness.make_block_return_pre_state(state, slot);
+    let (signed_block, mut state) = harness.make_block_return_pre_state(state, slot).await;
     let (mut block, signature) = signed_block.deconstruct();
     *block.slot_mut() = slot + Slot::new(1);
 
@@ -114,7 +116,9 @@ async fn invalid_parent_block_root() {
     let state = harness.get_current_state();
     let slot = state.slot();
 
-    let (signed_block, mut state) = harness.make_block_return_pre_state(state, slot + Slot::new(1));
+    let (signed_block, mut state) = harness
+        .make_block_return_pre_state(state, slot + Slot::new(1))
+        .await;
     let (mut block, signature) = signed_block.deconstruct();
     *block.parent_root_mut() = Hash256::from([0xAA; 32]);
 
@@ -145,7 +149,9 @@ async fn invalid_block_signature() {
 
     let state = harness.get_current_state();
     let slot = state.slot();
-    let (signed_block, mut state) = harness.make_block_return_pre_state(state, slot + Slot::new(1));
+    let (signed_block, mut state) = harness
+        .make_block_return_pre_state(state, slot + Slot::new(1))
+        .await;
     let (block, _) = signed_block.deconstruct();
 
     let result = per_block_processing(
@@ -174,9 +180,11 @@ async fn invalid_randao_reveal_signature() {
     let state = harness.get_current_state();
     let slot = state.slot();
 
-    let (signed_block, mut state) = harness.make_block_with_modifier(state, slot + 1, |block| {
-        *block.body_mut().randao_reveal_mut() = Signature::empty();
-    });
+    let (signed_block, mut state) = harness
+        .make_block_with_modifier(state, slot + 1, |block| {
+            *block.body_mut().randao_reveal_mut() = Signature::empty();
+        })
+        .await;
 
     let result = per_block_processing(
         &mut state,

diff --git a/consensus/types/src/payload.rs b/consensus/types/src/payload.rs
@@ -28,6 +28,8 @@ pub trait ExecPayload<T: EthSpec>:
     + Hash
     + TryFrom<ExecutionPayloadHeader<T>>
     + From<ExecutionPayload<T>>
+    + Send
+    + 'static
 {
     fn block_type() -> BlockType;
 

diff --git a/testing/state_transition_vectors/src/exit.rs b/testing/state_transition_vectors/src/exit.rs
@@ -50,11 +50,12 @@ impl ExitTest {
         let validator_index = self.validator_index;
         let exit_epoch = self.exit_epoch;
 
-        let (signed_block, state) =
-            harness.make_block_with_modifier(state.clone(), state.slot() + 1, |block| {
+        let (signed_block, state) = harness
+            .make_block_with_modifier(state.clone(), state.slot() + 1, |block| {
                 harness.add_voluntary_exit(block, validator_index, exit_epoch);
                 block_modifier(&harness, block);
-            });
+            })
+            .await;
         (signed_block, state)
     }