Electra devnet 1 process operations (#5994)

* Implement `process_consolidation_requests()`

* Finish Changes to `process_operations()`

* Fix Lint

* Fix test

* use electra_enabled()

beacon_node/execution_layer/src/engine_api.rs

@@ -20,7 +20,7 @@ use reqwest::StatusCode;
 use serde::{Deserialize, Serialize};
 use strum::IntoStaticStr;
 use superstruct::superstruct;
-use types::execution_payload::{ConsolidationRequests, DepositRequests, WithdrawalRequests};
+use types::execution_payload::{DepositRequests, WithdrawalRequests};
 pub use types::{
     Address, BeaconBlockRef, ConsolidationRequest, EthSpec, ExecutionBlockHash, ExecutionPayload,
     ExecutionPayloadHeader, ExecutionPayloadRef, FixedVector, ForkName, Hash256, Transactions,

beacon_node/execution_layer/src/engine_api/json_structures.rs

@@ -223,11 +223,7 @@ impl<E: EthSpec> From<ExecutionPayloadElectra<E>> for JsonExecutionPayloadV4<E>
                 .map(Into::into)
                 .collect::<Vec<_>>()
                 .into(),
-            consolidation_requests: payload
-                .consolidation_requests
-                .into_iter()
-                .collect::<Vec<_>>()
-                .into(),
+            consolidation_requests: payload.consolidation_requests,
         }
     }
 }
@@ -356,7 +352,7 @@ impl<E: EthSpec> From<JsonExecutionPayloadV4<E>> for ExecutionPayloadElectra<E>
                 .map(Into::into)
                 .collect::<Vec<_>>()
                 .into(),
-            consolidation_requests: payload.consolidation_requests.into(),
+            consolidation_requests: payload.consolidation_requests,
         }
     }
 }

beacon_node/execution_layer/src/lib.rs

@@ -2034,7 +2034,7 @@ impl<E: EthSpec> ExecutionLayer<E> {
                     excess_blob_gas: electra_block.excess_blob_gas,
                     deposit_requests,
                     withdrawal_requests,
-                    consolidation_requests: consolidation_requests,
+                    consolidation_requests,
                 })
             }
         };

beacon_node/http_api/tests/tests.rs

@@ -3210,7 +3210,7 @@ impl ApiTester {
                     .get_validator_aggregate_attestation_v2(
                         attestation.data().slot,
                         attestation.data().tree_hash_root(),
-                        attestation.committee_index(),
+                        attestation.committee_index().expect("committee index"),
                     )
                     .await
                     .unwrap()

consensus/state_processing/src/per_block_processing/errors.rs

@@ -89,46 +89,6 @@ pub enum BlockProcessingError {
         found: Hash256,
     },
     WithdrawalCredentialsInvalid,
-    TooManyPendingConsolidations {
-        consolidations: usize,
-        limit: usize,
-    },
-    ConsolidationChurnLimitTooLow {
-        churn_limit: u64,
-        minimum: u64,
-    },
-    MatchingSourceTargetConsolidation {
-        index: u64,
-    },
-    InactiveConsolidationSource {
-        index: u64,
-        current_epoch: Epoch,
-    },
-    InactiveConsolidationTarget {
-        index: u64,
-        current_epoch: Epoch,
-    },
-    SourceValidatorExiting {
-        index: u64,
-    },
-    TargetValidatorExiting {
-        index: u64,
-    },
-    FutureConsolidationEpoch {
-        current_epoch: Epoch,
-        consolidation_epoch: Epoch,
-    },
-    NoSourceExecutionWithdrawalCredential {
-        index: u64,
-    },
-    NoTargetExecutionWithdrawalCredential {
-        index: u64,
-    },
-    MismatchedWithdrawalCredentials {
-        source_address: Address,
-        target_address: Address,
-    },
-    InavlidConsolidationSignature,
     PendingAttestationInElectra,
 }
 

consensus/state_processing/src/per_block_processing/process_operations.rs

@@ -38,17 +38,15 @@ pub fn process_operations<E: EthSpec, Payload: AbstractExecPayload<E>>(
         process_bls_to_execution_changes(state, bls_to_execution_changes, verify_signatures, spec)?;
     }
 
-    if state.fork_name_unchecked() >= ForkName::Electra {
-        let requests = block_body.execution_payload()?.withdrawal_requests()?;
-        if let Some(requests) = requests {
-            process_execution_layer_withdrawal_requests(state, &requests, spec)?;
+    if state.fork_name_unchecked().electra_enabled() {
+        state.update_pubkey_cache()?;
+        if let Some(deposit_requests) = block_body.execution_payload()?.deposit_requests()? {
+            process_deposit_requests(state, &deposit_requests, spec)?;
         }
-        let receipts = block_body.execution_payload()?.deposit_requests()?;
-        if let Some(receipts) = receipts {
-            process_deposit_receipts(state, &receipts, spec)?;
+        if let Some(withdrawal_requests) = block_body.execution_payload()?.withdrawal_requests()? {
+            process_withdrawal_requests(state, &withdrawal_requests, spec)?;
         }
-        let consolidations = block_body.execution_payload()?.consolidation_requests()?;
-        if let Some(consolidations) = consolidations {
+        if let Some(consolidations) = block_body.execution_payload()?.consolidation_requests()? {
             process_consolidation_requests(state, &consolidations, spec)?;
         }
     }
@@ -373,10 +371,11 @@ pub fn process_deposits<E: EthSpec>(
 ) -> Result<(), BlockProcessingError> {
     // [Modified in Electra:EIP6110]
     // Disable former deposit mechanism once all prior deposits are processed
-    //
-    // If `deposit_requests_start_index` does not exist as a field on `state`, electra is disabled
-    // which means we always want to use the old check, so this field defaults to `u64::MAX`.
-    let eth1_deposit_index_limit = state.deposit_requests_start_index().unwrap_or(u64::MAX);
+    let deposit_requests_start_index = state.deposit_requests_start_index().unwrap_or(u64::MAX);
+    let eth1_deposit_index_limit = std::cmp::min(
+        deposit_requests_start_index,
+        state.eth1_data().deposit_count,
+    );
 
     if state.eth1_deposit_index() < eth1_deposit_index_limit {
         let expected_deposit_len = std::cmp::min(
@@ -530,7 +529,8 @@ pub fn apply_deposit<E: EthSpec>(
     Ok(())
 }
 
-pub fn process_execution_layer_withdrawal_requests<E: EthSpec>(
+// Make sure to build the pubkey cache before calling this function
+pub fn process_withdrawal_requests<E: EthSpec>(
     state: &mut BeaconState<E>,
     requests: &[WithdrawalRequest],
     spec: &ChainSpec,
@@ -547,13 +547,11 @@ pub fn process_execution_layer_withdrawal_requests<E: EthSpec>(
         }
 
         // Verify pubkey exists
-        let index_opt = state.get_validator_index(&request.validator_pubkey)?;
-        let Some(index) = index_opt else {
+        let Some(index) = state.pubkey_cache().get(&request.validator_pubkey) else {
             continue;
         };
 
         let validator = state.get_validator(index)?;
-
         // Verify withdrawal credentials
         let has_correct_credential = validator.has_execution_withdrawal_credential(spec);
         let is_correct_source_address = validator
@@ -627,32 +625,113 @@ pub fn process_execution_layer_withdrawal_requests<E: EthSpec>(
     Ok(())
 }
 
-pub fn process_deposit_receipts<E: EthSpec>(
+pub fn process_deposit_requests<E: EthSpec>(
     state: &mut BeaconState<E>,
-    receipts: &[DepositRequest],
+    deposit_requests: &[DepositRequest],
     spec: &ChainSpec,
 ) -> Result<(), BlockProcessingError> {
-    for receipt in receipts {
+    for request in deposit_requests {
         // Set deposit receipt start index
         if state.deposit_requests_start_index()? == spec.unset_deposit_requests_start_index {
-            *state.deposit_requests_start_index_mut()? = receipt.index
+            *state.deposit_requests_start_index_mut()? = request.index
         }
         let deposit_data = DepositData {
-            pubkey: receipt.pubkey,
-            withdrawal_credentials: receipt.withdrawal_credentials,
-            amount: receipt.amount,
-            signature: receipt.signature.clone().into(),
+            pubkey: request.pubkey,
+            withdrawal_credentials: request.withdrawal_credentials,
+            amount: request.amount,
+            signature: request.signature.clone().into(),
         };
         apply_deposit(state, deposit_data, None, false, spec)?
     }
 
     Ok(())
 }
 
+// Make sure to build the pubkey cache before calling this function
 pub fn process_consolidation_requests<E: EthSpec>(
     state: &mut BeaconState<E>,
     consolidation_requests: &[ConsolidationRequest],
     spec: &ChainSpec,
 ) -> Result<(), BlockProcessingError> {
-    todo!("implement process_consolidation_requests");
+    for request in consolidation_requests {
+        process_consolidation_request(state, request, spec)?;
+    }
+
+    Ok(())
+}
+
+pub fn process_consolidation_request<E: EthSpec>(
+    state: &mut BeaconState<E>,
+    consolidation_request: &ConsolidationRequest,
+    spec: &ChainSpec,
+) -> Result<(), BlockProcessingError> {
+    // If the pending consolidations queue is full, consolidation requests are ignored
+    if state.pending_consolidations()?.len() == E::PendingConsolidationsLimit::to_usize() {
+        return Ok(());
+    }
+    // If there is too little available consolidation churn limit, consolidation requests are ignored
+    if state.get_consolidation_churn_limit(spec)? <= spec.min_activation_balance {
+        return Ok(());
+    }
+
+    let Some(source_index) = state
+        .pubkey_cache()
+        .get(&consolidation_request.source_pubkey)
+    else {
+        // source validator doesn't exist
+        return Ok(());
+    };
+    let Some(target_index) = state
+        .pubkey_cache()
+        .get(&consolidation_request.target_pubkey)
+    else {
+        // target validator doesn't exist
+        return Ok(());
+    };
+    // Verify that source != target, so a consolidation cannot be used as an exit.
+    if source_index == target_index {
+        return Ok(());
+    }
+
+    let source_validator = state.get_validator(source_index)?;
+    // Verify the source withdrawal credentials
+    if let Some(withdrawal_address) = source_validator.get_execution_withdrawal_address(spec) {
+        if withdrawal_address != consolidation_request.source_address {
+            return Ok(());
+        }
+    } else {
+        // Source doen't have execution withdrawal credentials
+        return Ok(());
+    }
+
+    let target_validator = state.get_validator(target_index)?;
+    // Verify the target has execution withdrawal credentials
+    if !target_validator.has_execution_withdrawal_credential(spec) {
+        return Ok(());
+    }
+
+    // Verify the source and target are active
+    let current_epoch = state.current_epoch();
+    if !source_validator.is_active_at(current_epoch)
+        || !target_validator.is_active_at(current_epoch)
+    {
+        return Ok(());
+    }
+    // Verify exits for source and target have not been initiated
+    if source_validator.exit_epoch != spec.far_future_epoch
+        || target_validator.exit_epoch != spec.far_future_epoch
+    {
+        return Ok(());
+    }
+
+    // Initiate source validator exit and append pending consolidation
+    initiate_validator_exit(state, source_index, spec)?;
+    state
+        .pending_consolidations_mut()?
+        .push(PendingConsolidation {
+            source_index: source_index as u64,
+            target_index: target_index as u64,
+        })?;
+
+    Ok(())
 }

testing/ef_tests/src/cases/operations.rs

@@ -7,8 +7,7 @@ use ssz::Decode;
 use state_processing::common::update_progressive_balances_cache::initialize_progressive_balances_cache;
 use state_processing::epoch_cache::initialize_epoch_cache;
 use state_processing::per_block_processing::process_operations::{
-    process_consolidation_requests, process_deposit_receipts,
-    process_execution_layer_withdrawal_requests,
+    process_deposit_requests, process_withdrawal_requests,
 };
 use state_processing::{
     per_block_processing::{
@@ -464,7 +463,7 @@ impl<E: EthSpec> Operation<E> for WithdrawalRequest {
         spec: &ChainSpec,
         _extra: &Operations<E, Self>,
     ) -> Result<(), BlockProcessingError> {
-        process_execution_layer_withdrawal_requests(state, &[self.clone()], spec)
+        process_withdrawal_requests(state, &[self.clone()], spec)
     }
 }
 
@@ -487,7 +486,7 @@ impl<E: EthSpec> Operation<E> for DepositRequest {
         spec: &ChainSpec,
         _extra: &Operations<E, Self>,
     ) -> Result<(), BlockProcessingError> {
-        process_deposit_receipts(state, &[self.clone()], spec)
+        process_deposit_requests(state, &[self.clone()], spec)
     }
 }