Merge branch 'capella' into unstable

beacon_node/http_api/Cargo.toml

@@ -37,6 +37,7 @@ sysinfo = "0.26.5"
 system_health = { path = "../../common/system_health" }
 directory = { path = "../../common/directory" }
 eth2_serde_utils = "0.1.1"
+operation_pool = { path = "../operation_pool" }
 
 [dev-dependencies]
 store = { path = "../store" }
@@ -46,6 +47,7 @@ logging = { path = "../../common/logging" }
 serde_json = "1.0.58"
 proto_array = { path = "../../consensus/proto_array" }
 unused_port = {path = "../../common/unused_port"}
+genesis = { path = "../genesis" }
 
 [[test]]
 name = "bn_http_api_tests"

beacon_node/http_api/src/block_rewards.rs

@@ -4,7 +4,7 @@ use lru::LruCache;
 use slog::{debug, warn, Logger};
 use state_processing::BlockReplayer;
 use std::sync::Arc;
-use types::BlindedBeaconBlock;
+use types::beacon_block::BlindedBeaconBlock;
 use warp_utils::reject::{
     beacon_chain_error, beacon_state_error, custom_bad_request, custom_server_error,
 };

beacon_node/http_api/src/lib.rs

@@ -36,6 +36,7 @@ use eth2::types::{
 use lighthouse_network::{types::SyncState, EnrExt, NetworkGlobals, PeerId, PubsubMessage};
 use lighthouse_version::version_with_platform;
 use network::{NetworkMessage, NetworkSenders, ValidatorSubscriptionMessage};
+use operation_pool::ReceivedPreCapella;
 use parking_lot::RwLock;
 use serde::{Deserialize, Serialize};
 use slog::{crit, debug, error, info, warn, Logger};
@@ -56,9 +57,9 @@ use types::{
     Attestation, AttestationData, AttesterSlashing, BeaconStateError, BlindedPayload,
     CommitteeCache, ConfigAndPreset, Epoch, EthSpec, ForkName, FullPayload,
     ProposerPreparationData, ProposerSlashing, RelativeEpoch, SignedAggregateAndProof,
-    SignedBeaconBlock, SignedBlindedBeaconBlock, SignedContributionAndProof,
-    SignedValidatorRegistrationData, SignedVoluntaryExit, Slot, SyncCommitteeMessage,
-    SyncContributionData,
+    SignedBeaconBlock, SignedBlindedBeaconBlock, SignedBlsToExecutionChange,
+    SignedContributionAndProof, SignedValidatorRegistrationData, SignedVoluntaryExit, Slot,
+    SyncCommitteeMessage, SyncContributionData,
 };
 use version::{
     add_consensus_version_header, execution_optimistic_fork_versioned_response,
@@ -1122,7 +1123,9 @@ pub fn serve<T: BeaconChainTypes>(
              chain: Arc<BeaconChain<T>>,
              network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>,
              log: Logger| async move {
-                publish_blocks::publish_block(None, block, chain, &network_tx, log)
+                // need to have cached the blob sidecar somewhere in the beacon chain
+                // to publish
+                publish_blocks::publish_block(None, block, None, chain, &network_tx, log)
                     .await
                     .map(|()| warp::reply())
             },
@@ -1654,6 +1657,109 @@ pub fn serve<T: BeaconChainTypes>(
             },
         );
 
+    // GET beacon/pool/bls_to_execution_changes
+    let get_beacon_pool_bls_to_execution_changes = beacon_pool_path
+        .clone()
+        .and(warp::path("bls_to_execution_changes"))
+        .and(warp::path::end())
+        .and_then(|chain: Arc<BeaconChain<T>>| {
+            blocking_json_task(move || {
+                let address_changes = chain.op_pool.get_all_bls_to_execution_changes();
+                Ok(api_types::GenericResponse::from(address_changes))
+            })
+        });
+
+    // POST beacon/pool/bls_to_execution_changes
+    let post_beacon_pool_bls_to_execution_changes = beacon_pool_path
+        .clone()
+        .and(warp::path("bls_to_execution_changes"))
+        .and(warp::path::end())
+        .and(warp::body::json())
+        .and(network_tx_filter.clone())
+        .and(log_filter.clone())
+        .and_then(
+            |chain: Arc<BeaconChain<T>>,
+             address_changes: Vec<SignedBlsToExecutionChange>,
+             network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>,
+             log: Logger| {
+                blocking_json_task(move || {
+                    let mut failures = vec![];
+
+                    for (index, address_change) in address_changes.into_iter().enumerate() {
+                        let validator_index = address_change.message.validator_index;
+
+                        match chain.verify_bls_to_execution_change_for_http_api(address_change) {
+                            Ok(ObservationOutcome::New(verified_address_change)) => {
+                                let validator_index =
+                                    verified_address_change.as_inner().message.validator_index;
+                                let address = verified_address_change
+                                    .as_inner()
+                                    .message
+                                    .to_execution_address;
+
+                                // New to P2P *and* op pool, gossip immediately if post-Capella.
+                                let received_pre_capella = if chain.current_slot_is_post_capella().unwrap_or(false) {
+                                    ReceivedPreCapella::No
+                                } else {
+                                    ReceivedPreCapella::Yes
+                                };
+                                if matches!(received_pre_capella, ReceivedPreCapella::No) {
+                                    publish_pubsub_message(
+                                        &network_tx,
+                                        PubsubMessage::BlsToExecutionChange(Box::new(
+                                            verified_address_change.as_inner().clone(),
+                                        )),
+                                    )?;
+                                }
+
+                                // Import to op pool (may return `false` if there's a race).
+                                let imported =
+                                    chain.import_bls_to_execution_change(verified_address_change, received_pre_capella);
+
+                                info!(
+                                    log,
+                                    "Processed BLS to execution change";
+                                    "validator_index" => validator_index,
+                                    "address" => ?address,
+                                    "published" => matches!(received_pre_capella, ReceivedPreCapella::No),
+                                    "imported" => imported,
+                                );
+                            }
+                            Ok(ObservationOutcome::AlreadyKnown) => {
+                                debug!(
+                                    log,
+                                    "BLS to execution change already known";
+                                    "validator_index" => validator_index,
+                                );
+                            }
+                            Err(e) => {
+                                warn!(
+                                    log,
+                                    "Invalid BLS to execution change";
+                                    "validator_index" => validator_index,
+                                    "reason" => ?e,
+                                    "source" => "HTTP",
+                                );
+                                failures.push(api_types::Failure::new(
+                                    index,
+                                    format!("invalid: {e:?}"),
+                                ));
+                            }
+                        }
+                    }
+
+                    if failures.is_empty() {
+                        Ok(())
+                    } else {
+                        Err(warp_utils::reject::indexed_bad_request(
+                            "some BLS to execution changes failed to verify".into(),
+                            failures,
+                        ))
+                    }
+                })
+            },
+        );
+
     // GET beacon/deposit_snapshot
     let get_beacon_deposit_snapshot = eth_v1
         .and(warp::path("beacon"))
@@ -3470,6 +3576,7 @@ pub fn serve<T: BeaconChainTypes>(
                 .or(get_beacon_pool_attester_slashings.boxed())
                 .or(get_beacon_pool_proposer_slashings.boxed())
                 .or(get_beacon_pool_voluntary_exits.boxed())
+                .or(get_beacon_pool_bls_to_execution_changes.boxed())
                 .or(get_beacon_deposit_snapshot.boxed())
                 .or(get_beacon_rewards_blocks.boxed())
                 .or(get_config_fork_schedule.boxed())
@@ -3523,6 +3630,7 @@ pub fn serve<T: BeaconChainTypes>(
                 .or(post_beacon_pool_proposer_slashings.boxed())
                 .or(post_beacon_pool_voluntary_exits.boxed())
                 .or(post_beacon_pool_sync_committees.boxed())
+                .or(post_beacon_pool_bls_to_execution_changes.boxed())
                 .or(post_beacon_rewards_attestations.boxed())
                 .or(post_beacon_rewards_sync_committee.boxed())
                 .or(post_validator_duties_attester.boxed())

beacon_node/http_api/src/metrics.rs

@@ -41,4 +41,16 @@ lazy_static::lazy_static! {
         "http_api_block_published_very_late_total",
         "The count of times a block was published beyond the attestation deadline"
     );
+    pub static ref HTTP_API_BLOB_BROADCAST_DELAY_TIMES: Result<Histogram> = try_create_histogram(
+        "http_api_blob_broadcast_delay_times",
+        "Time between start of the slot and when the blob was broadcast"
+    );
+    pub static ref HTTP_API_BLOB_PUBLISHED_LATE_TOTAL: Result<IntCounter> = try_create_int_counter(
+        "http_api_blob_published_late_total",
+        "The count of times a blob was published beyond more than half way to the attestation deadline"
+    );
+    pub static ref HTTP_API_BLOB_PUBLISHED_VERY_LATE_TOTAL: Result<IntCounter> = try_create_int_counter(
+        "http_api_blob_published_very_late_total",
+        "The count of times a blob was published beyond the attestation deadline"
+    );
 }

beacon_node/http_api/src/publish_blocks.rs

@@ -3,7 +3,7 @@ use beacon_chain::validator_monitor::{get_block_delay_ms, timestamp_now};
 use beacon_chain::{
     BeaconChain, BeaconChainTypes, BlockError, CountUnrealized, NotifyExecutionLayer,
 };
-use lighthouse_network::PubsubMessage;
+use lighthouse_network::{PubsubMessage, SignedBeaconBlockAndBlobsSidecar};
 use network::NetworkMessage;
 use slog::{error, info, warn, Logger};
 use slot_clock::SlotClock;
@@ -11,8 +11,8 @@ use std::sync::Arc;
 use tokio::sync::mpsc::UnboundedSender;
 use tree_hash::TreeHash;
 use types::{
-    BlindedPayload, ExecPayload, ExecutionBlockHash, ExecutionPayload, FullPayload, Hash256,
-    SignedBeaconBlock,
+    AbstractExecPayload, BlindedPayload, BlobsSidecar, EthSpec, ExecPayload, ExecutionBlockHash,
+    FullPayload, Hash256, SignedBeaconBlock,
 };
 use warp::Rejection;
 
@@ -20,6 +20,7 @@ use warp::Rejection;
 pub async fn publish_block<T: BeaconChainTypes>(
     block_root: Option<Hash256>,
     block: Arc<SignedBeaconBlock<T::EthSpec>>,
+    blobs_sidecar: Option<Arc<BlobsSidecar<T::EthSpec>>>,
     chain: Arc<BeaconChain<T>>,
     network_tx: &UnboundedSender<NetworkMessage<T::EthSpec>>,
     log: Logger,
@@ -28,7 +29,24 @@ pub async fn publish_block<T: BeaconChainTypes>(
 
     // Send the block, regardless of whether or not it is valid. The API
     // specification is very clear that this is the desired behaviour.
-    crate::publish_pubsub_message(network_tx, PubsubMessage::BeaconBlock(block.clone()))?;
+
+    let message = match &*block {
+        SignedBeaconBlock::Eip4844(block) => {
+            if let Some(sidecar) = blobs_sidecar {
+                PubsubMessage::BeaconBlockAndBlobsSidecars(Arc::new(
+                    SignedBeaconBlockAndBlobsSidecar {
+                        beacon_block: block.clone(),
+                        blobs_sidecar: (*sidecar).clone(),
+                    },
+                ))
+            } else {
+                //TODO(pawan): return an empty sidecar instead
+                return Err(warp_utils::reject::broadcast_without_import(String::new()));
+            }
+        }
+        _ => PubsubMessage::BeaconBlock(block.clone()),
+    };
+    crate::publish_pubsub_message(network_tx, message)?;
 
     // Determine the delay after the start of the slot, register it with metrics.
     let delay = get_block_delay_ms(seen_timestamp, block.message(), &chain.slot_clock);
@@ -142,6 +160,7 @@ pub async fn publish_blinded_block<T: BeaconChainTypes>(
     publish_block::<T>(
         Some(block_root),
         Arc::new(full_block),
+        None,
         chain,
         network_tx,
         log,
@@ -165,12 +184,22 @@ async fn reconstruct_block<T: BeaconChainTypes>(
 
         // If the execution block hash is zero, use an empty payload.
         let full_payload = if payload_header.block_hash() == ExecutionBlockHash::zero() {
-            ExecutionPayload::default()
+            FullPayload::default_at_fork(
+                chain
+                    .spec
+                    .fork_name_at_epoch(block.slot().epoch(T::EthSpec::slots_per_epoch())),
+            )
+            .map_err(|e| {
+                warp_utils::reject::custom_server_error(format!(
+                    "Default payload construction error: {e:?}"
+                ))
+            })?
+            .into()
             // If we already have an execution payload with this transactions root cached, use it.
         } else if let Some(cached_payload) =
             el.get_payload_by_root(&payload_header.tree_hash_root())
         {
-            info!(log, "Reconstructing a full block using a local payload"; "block_hash" => ?cached_payload.block_hash);
+            info!(log, "Reconstructing a full block using a local payload"; "block_hash" => ?cached_payload.block_hash());
             cached_payload
             // Otherwise, this means we are attempting a blind block proposal.
         } else {
@@ -183,7 +212,7 @@ async fn reconstruct_block<T: BeaconChainTypes>(
                         e
                     ))
                 })?;
-            info!(log, "Successfully published a block to the builder network"; "block_hash" => ?full_payload.block_hash);
+            info!(log, "Successfully published a block to the builder network"; "block_hash" => ?full_payload.block_hash());
             full_payload
         };
 

beacon_node/http_api/src/version.rs

@@ -1,9 +1,9 @@
-use crate::api_types::{
-    EndpointVersion, ExecutionOptimisticForkVersionedResponse, ForkVersionedResponse,
-};
+use crate::api_types::EndpointVersion;
 use eth2::CONSENSUS_VERSION_HEADER;
 use serde::Serialize;
-use types::{ForkName, InconsistentFork};
+use types::{
+    ExecutionOptimisticForkVersionedResponse, ForkName, ForkVersionedResponse, InconsistentFork,
+};
 use warp::reply::{self, Reply, WithHeader};
 
 pub const V1: EndpointVersion = EndpointVersion(1);

beacon_node/http_api/tests/common.rs

@@ -1,5 +1,7 @@
 use beacon_chain::{
-    test_utils::{BeaconChainHarness, BoxedMutator, EphemeralHarnessType},
+    test_utils::{
+        BeaconChainHarness, BoxedMutator, Builder as HarnessBuilder, EphemeralHarnessType,
+    },
     BeaconChain, BeaconChainTypes,
 };
 use directory::DEFAULT_ROOT_DIR;
@@ -55,25 +57,39 @@ pub struct ApiServer<E: EthSpec, SFut: Future<Output = ()>> {
     pub external_peer_id: PeerId,
 }
 
+type Initializer<E> = Box<
+    dyn FnOnce(HarnessBuilder<EphemeralHarnessType<E>>) -> HarnessBuilder<EphemeralHarnessType<E>>,
+>;
 type Mutator<E> = BoxedMutator<E, MemoryStore<E>, MemoryStore<E>>;
 
 impl<E: EthSpec> InteractiveTester<E> {
     pub async fn new(spec: Option<ChainSpec>, validator_count: usize) -> Self {
-        Self::new_with_mutator(spec, validator_count, None).await
+        Self::new_with_initializer_and_mutator(spec, validator_count, None, None).await
     }
 
-    pub async fn new_with_mutator(
+    pub async fn new_with_initializer_and_mutator(
         spec: Option<ChainSpec>,
         validator_count: usize,
+        initializer: Option<Initializer<E>>,
         mutator: Option<Mutator<E>>,
     ) -> Self {
         let mut harness_builder = BeaconChainHarness::builder(E::default())
             .spec_or_default(spec)
-            .deterministic_keypairs(validator_count)
             .logger(test_logger())
-            .mock_execution_layer()
-            .fresh_ephemeral_store();
+            .mock_execution_layer();
 
+        harness_builder = if let Some(initializer) = initializer {
+            // Apply custom initialization provided by the caller.
+            initializer(harness_builder)
+        } else {
+            // Apply default initial configuration.
+            harness_builder
+                .deterministic_keypairs(validator_count)
+                .fresh_ephemeral_store()
+        };
+
+        // Add a mutator for the beacon chain builder which will be called in
+        // `HarnessBuilder::build`.
         if let Some(mutator) = mutator {
             harness_builder = harness_builder.initial_mutator(mutator);
         }

beacon_node/http_api/tests/fork_tests.rs

@@ -1,8 +1,16 @@
 //! Tests for API behaviour across fork boundaries.
 use crate::common::*;
-use beacon_chain::{test_utils::RelativeSyncCommittee, StateSkipConfig};
-use eth2::types::{StateId, SyncSubcommittee};
-use types::{ChainSpec, Epoch, EthSpec, MinimalEthSpec, Slot};
+use beacon_chain::{
+    test_utils::{RelativeSyncCommittee, DEFAULT_ETH1_BLOCK_HASH, HARNESS_GENESIS_TIME},
+    StateSkipConfig,
+};
+use eth2::types::{IndexedErrorMessage, StateId, SyncSubcommittee};
+use genesis::{bls_withdrawal_credentials, interop_genesis_state_with_withdrawal_credentials};
+use std::collections::HashSet;
+use types::{
+    test_utils::{generate_deterministic_keypair, generate_deterministic_keypairs},
+    Address, ChainSpec, Epoch, EthSpec, Hash256, MinimalEthSpec, Slot,
+};
 
 type E = MinimalEthSpec;
 
@@ -12,6 +20,14 @@ fn altair_spec(altair_fork_epoch: Epoch) -> ChainSpec {
     spec
 }
 
+fn capella_spec(capella_fork_epoch: Epoch) -> ChainSpec {
+    let mut spec = E::default_spec();
+    spec.altair_fork_epoch = Some(Epoch::new(0));
+    spec.bellatrix_fork_epoch = Some(Epoch::new(0));
+    spec.capella_fork_epoch = Some(capella_fork_epoch);
+    spec
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn sync_committee_duties_across_fork() {
     let validator_count = E::sync_committee_size();
@@ -307,3 +323,219 @@ async fn sync_committee_indices_across_fork() {
         );
     }
 }
+
+/// Assert that an HTTP API error has the given status code and indexed errors for the given indices.
+fn assert_server_indexed_error(error: eth2::Error, status_code: u16, indices: Vec<usize>) {
+    let eth2::Error::ServerIndexedMessage(IndexedErrorMessage {
+        code,
+        failures,
+        ..
+    }) = error else {
+        panic!("wrong error, expected ServerIndexedMessage, got: {error:?}")
+    };
+    assert_eq!(code, status_code);
+    assert_eq!(failures.len(), indices.len());
+    for (index, failure) in indices.into_iter().zip(failures) {
+        assert_eq!(failure.index, index as u64);
+    }
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn bls_to_execution_changes_update_all_around_capella_fork() {
+    let validator_count = 128;
+    let fork_epoch = Epoch::new(2);
+    let spec = capella_spec(fork_epoch);
+    let max_bls_to_execution_changes = E::max_bls_to_execution_changes();
+
+    // Use a genesis state with entirely BLS withdrawal credentials.
+    // Offset keypairs by `validator_count` to create keys distinct from the signing keys.
+    let validator_keypairs = generate_deterministic_keypairs(validator_count);
+    let withdrawal_keypairs = (0..validator_count)
+        .map(|i| Some(generate_deterministic_keypair(i + validator_count)))
+        .collect::<Vec<_>>();
+    let withdrawal_credentials = withdrawal_keypairs
+        .iter()
+        .map(|keypair| bls_withdrawal_credentials(&keypair.as_ref().unwrap().pk, &spec))
+        .collect::<Vec<_>>();
+    let genesis_state = interop_genesis_state_with_withdrawal_credentials(
+        &validator_keypairs,
+        &withdrawal_credentials,
+        HARNESS_GENESIS_TIME,
+        Hash256::from_slice(DEFAULT_ETH1_BLOCK_HASH),
+        None,
+        &spec,
+    )
+    .unwrap();
+
+    let tester = InteractiveTester::<E>::new_with_initializer_and_mutator(
+        Some(spec.clone()),
+        validator_count,
+        Some(Box::new(|harness_builder| {
+            harness_builder
+                .keypairs(validator_keypairs)
+                .withdrawal_keypairs(withdrawal_keypairs)
+                .genesis_state_ephemeral_store(genesis_state)
+        })),
+        None,
+    )
+    .await;
+    let harness = &tester.harness;
+    let client = &tester.client;
+
+    let all_validators = harness.get_all_validators();
+    let all_validators_u64 = all_validators.iter().map(|x| *x as u64).collect::<Vec<_>>();
+
+    // Create a bunch of valid address changes.
+    let valid_address_changes = all_validators_u64
+        .iter()
+        .map(|&validator_index| {
+            harness.make_bls_to_execution_change(
+                validator_index,
+                Address::from_low_u64_be(validator_index),
+            )
+        })
+        .collect::<Vec<_>>();
+
+    // Address changes which conflict with `valid_address_changes` on the address chosen.
+    let conflicting_address_changes = all_validators_u64
+        .iter()
+        .map(|&validator_index| {
+            harness.make_bls_to_execution_change(
+                validator_index,
+                Address::from_low_u64_be(validator_index + 1),
+            )
+        })
+        .collect::<Vec<_>>();
+
+    // Address changes signed with the wrong key.
+    let wrong_key_address_changes = all_validators_u64
+        .iter()
+        .map(|&validator_index| {
+            // Use the correct pubkey.
+            let pubkey = &harness.get_withdrawal_keypair(validator_index).pk;
+            // And the wrong secret key.
+            let secret_key = &harness
+                .get_withdrawal_keypair((validator_index + 1) % validator_count as u64)
+                .sk;
+            harness.make_bls_to_execution_change_with_keys(
+                validator_index,
+                Address::from_low_u64_be(validator_index),
+                pubkey,
+                secret_key,
+            )
+        })
+        .collect::<Vec<_>>();
+
+    // Submit some changes before Capella. Just enough to fill two blocks.
+    let num_pre_capella = validator_count / 4;
+    let blocks_filled_pre_capella = 2;
+    assert_eq!(
+        num_pre_capella,
+        blocks_filled_pre_capella * max_bls_to_execution_changes
+    );
+
+    client
+        .post_beacon_pool_bls_to_execution_changes(&valid_address_changes[..num_pre_capella])
+        .await
+        .unwrap();
+
+    let expected_received_pre_capella_messages = valid_address_changes[..num_pre_capella].to_vec();
+
+    // Conflicting changes for the same validators should all fail.
+    let error = client
+        .post_beacon_pool_bls_to_execution_changes(&conflicting_address_changes[..num_pre_capella])
+        .await
+        .unwrap_err();
+    assert_server_indexed_error(error, 400, (0..num_pre_capella).collect());
+
+    // Re-submitting the same changes should be accepted.
+    client
+        .post_beacon_pool_bls_to_execution_changes(&valid_address_changes[..num_pre_capella])
+        .await
+        .unwrap();
+
+    // Invalid changes signed with the wrong keys should all be rejected without affecting the seen
+    // indices filters (apply ALL of them).
+    let error = client
+        .post_beacon_pool_bls_to_execution_changes(&wrong_key_address_changes)
+        .await
+        .unwrap_err();
+    assert_server_indexed_error(error, 400, all_validators.clone());
+
+    // Advance to right before Capella.
+    let capella_slot = fork_epoch.start_slot(E::slots_per_epoch());
+    harness.extend_to_slot(capella_slot - 1).await;
+    assert_eq!(harness.head_slot(), capella_slot - 1);
+
+    assert_eq!(
+        harness
+            .chain
+            .op_pool
+            .get_bls_to_execution_changes_received_pre_capella(
+                &harness.chain.head_snapshot().beacon_state,
+                &spec,
+            )
+            .into_iter()
+            .collect::<HashSet<_>>(),
+        HashSet::from_iter(expected_received_pre_capella_messages.into_iter()),
+        "all pre-capella messages should be queued for capella broadcast"
+    );
+
+    // Add Capella blocks which should be full of BLS to execution changes.
+    for i in 0..validator_count / max_bls_to_execution_changes {
+        let head_block_root = harness.extend_slots(1).await;
+        let head_block = harness
+            .chain
+            .get_block(&head_block_root)
+            .await
+            .unwrap()
+            .unwrap();
+
+        let bls_to_execution_changes = head_block
+            .message()
+            .body()
+            .bls_to_execution_changes()
+            .unwrap();
+
+        // Block should be full.
+        assert_eq!(
+            bls_to_execution_changes.len(),
+            max_bls_to_execution_changes,
+            "block not full on iteration {i}"
+        );
+
+        // Included changes should be the ones from `valid_address_changes` in any order.
+        for address_change in bls_to_execution_changes.iter() {
+            assert!(valid_address_changes.contains(address_change));
+        }
+
+        // After the initial 2 blocks, add the rest of the changes using a large
+        // request containing all the valid, all the conflicting and all the invalid.
+        // Despite the invalid and duplicate messages, the new ones should still get picked up by
+        // the pool.
+        if i == blocks_filled_pre_capella - 1 {
+            let all_address_changes: Vec<_> = [
+                valid_address_changes.clone(),
+                conflicting_address_changes.clone(),
+                wrong_key_address_changes.clone(),
+            ]
+            .concat();
+
+            let error = client
+                .post_beacon_pool_bls_to_execution_changes(&all_address_changes)
+                .await
+                .unwrap_err();
+            assert_server_indexed_error(
+                error,
+                400,
+                (validator_count..3 * validator_count).collect(),
+            );
+        }
+    }
+
+    // Eventually all validators should have eth1 withdrawal credentials.
+    let head_state = harness.get_current_state();
+    for validator in head_state.validators() {
+        assert!(validator.has_eth1_withdrawal_credential(&spec));
+    }
+}

beacon_node/http_api/tests/interactive_tests.rs

@@ -5,7 +5,7 @@ use beacon_chain::{
     test_utils::{AttestationStrategy, BlockStrategy},
 };
 use eth2::types::DepositContractData;
-use execution_layer::{ForkChoiceState, PayloadAttributes};
+use execution_layer::{ForkchoiceState, PayloadAttributes};
 use parking_lot::Mutex;
 use slot_clock::SlotClock;
 use state_processing::state_advance::complete_state_advance;
@@ -55,7 +55,7 @@ struct ForkChoiceUpdates {
 #[derive(Debug, Clone)]
 struct ForkChoiceUpdateMetadata {
     received_at: Duration,
-    state: ForkChoiceState,
+    state: ForkchoiceState,
     payload_attributes: Option<PayloadAttributes>,
 }
 
@@ -86,7 +86,7 @@ impl ForkChoiceUpdates {
                     .payload_attributes
                     .as_ref()
                     .map_or(false, |payload_attributes| {
-                        payload_attributes.timestamp == proposal_timestamp
+                        payload_attributes.timestamp() == proposal_timestamp
                     })
             })
             .cloned()
@@ -278,9 +278,10 @@ pub async fn proposer_boost_re_org_test(
     let num_empty_votes = Some(attesters_per_slot * percent_empty_votes / 100);
     let num_head_votes = Some(attesters_per_slot * percent_head_votes / 100);
 
-    let tester = InteractiveTester::<E>::new_with_mutator(
+    let tester = InteractiveTester::<E>::new_with_initializer_and_mutator(
         Some(spec),
         validator_count,
+        None,
         Some(Box::new(move |builder| {
             builder
                 .proposer_re_org_threshold(Some(ReOrgThreshold(re_org_threshold)))
@@ -342,7 +343,7 @@ pub async fn proposer_boost_re_org_test(
         .lock()
         .set_forkchoice_updated_hook(Box::new(move |state, payload_attributes| {
             let received_at = chain_inner.slot_clock.now_duration().unwrap();
-            let state = ForkChoiceState::from(state);
+            let state = ForkchoiceState::from(state);
             let payload_attributes = payload_attributes.map(Into::into);
             let update = ForkChoiceUpdateMetadata {
                 received_at,
@@ -521,16 +522,20 @@ pub async fn proposer_boost_re_org_test(
 
     if !misprediction {
         assert_eq!(
-            lookahead, payload_lookahead,
+            lookahead,
+            payload_lookahead,
             "lookahead={lookahead:?}, timestamp={}, prev_randao={:?}",
-            payload_attribs.timestamp, payload_attribs.prev_randao,
+            payload_attribs.timestamp(),
+            payload_attribs.prev_randao(),
         );
     } else {
         // On a misprediction we issue the first fcU 500ms before creating a block!
         assert_eq!(
-            lookahead, fork_choice_lookahead,
+            lookahead,
+            fork_choice_lookahead,
             "timestamp={}, prev_randao={:?}",
-            payload_attribs.timestamp, payload_attribs.prev_randao,
+            payload_attribs.timestamp(),
+            payload_attribs.prev_randao(),
         );
     }
 }
@@ -540,7 +545,7 @@ pub async fn proposer_boost_re_org_test(
 pub async fn fork_choice_before_proposal() {
     // Validator count needs to be at least 32 or proposer boost gets set to 0 when computing
     // `validator_count // 32`.
-    let validator_count = 32;
+    let validator_count = 64;
     let all_validators = (0..validator_count).collect::<Vec<_>>();
     let num_initial: u64 = 31;
 

beacon_node/http_api/tests/tests.rs

@@ -11,9 +11,11 @@ use eth2::{
     types::{BlockId as CoreBlockId, StateId as CoreStateId, *},
     BeaconNodeHttpClient, Error, StatusCode, Timeouts,
 };
-use execution_layer::test_utils::Operation;
 use execution_layer::test_utils::TestingBuilder;
 use execution_layer::test_utils::DEFAULT_BUILDER_THRESHOLD_WEI;
+use execution_layer::test_utils::{
+    Operation, DEFAULT_BUILDER_PAYLOAD_VALUE_WEI, DEFAULT_MOCK_EL_PAYLOAD_VALUE_WEI,
+};
 use futures::stream::{Stream, StreamExt};
 use futures::FutureExt;
 use http_api::{BlockId, StateId};
@@ -22,6 +24,7 @@ use network::NetworkReceivers;
 use proto_array::ExecutionStatus;
 use sensitive_url::SensitiveUrl;
 use slot_clock::SlotClock;
+use state_processing::per_block_processing::get_expected_withdrawals;
 use state_processing::per_slot_processing;
 use std::convert::TryInto;
 use std::sync::Arc;
@@ -72,38 +75,53 @@ struct ApiTester {
     mock_builder: Option<Arc<TestingBuilder<E>>>,
 }
 
+struct ApiTesterConfig {
+    spec: ChainSpec,
+    builder_threshold: Option<u128>,
+}
+
+impl Default for ApiTesterConfig {
+    fn default() -> Self {
+        let mut spec = E::default_spec();
+        spec.shard_committee_period = 2;
+        Self {
+            spec,
+            builder_threshold: None,
+        }
+    }
+}
+
 impl ApiTester {
     pub async fn new() -> Self {
         // This allows for testing voluntary exits without building out a massive chain.
-        let mut spec = E::default_spec();
-        spec.shard_committee_period = 2;
-        Self::new_from_spec(spec).await
+        Self::new_from_config(ApiTesterConfig::default()).await
     }
 
     pub async fn new_with_hard_forks(altair: bool, bellatrix: bool) -> Self {
-        let mut spec = E::default_spec();
-        spec.shard_committee_period = 2;
+        let mut config = ApiTesterConfig::default();
         // Set whether the chain has undergone each hard fork.
         if altair {
-            spec.altair_fork_epoch = Some(Epoch::new(0));
+            config.spec.altair_fork_epoch = Some(Epoch::new(0));
         }
         if bellatrix {
-            spec.bellatrix_fork_epoch = Some(Epoch::new(0));
+            config.spec.bellatrix_fork_epoch = Some(Epoch::new(0));
         }
-        Self::new_from_spec(spec).await
+        Self::new_from_config(config).await
     }
 
-    pub async fn new_from_spec(spec: ChainSpec) -> Self {
+    pub async fn new_from_config(config: ApiTesterConfig) -> Self {
         // Get a random unused port
+        let spec = config.spec;
         let port = unused_port::unused_tcp_port().unwrap();
         let beacon_url = SensitiveUrl::parse(format!("http://127.0.0.1:{port}").as_str()).unwrap();
 
         let harness = Arc::new(
             BeaconChainHarness::builder(MainnetEthSpec)
                 .spec(spec.clone())
+                .logger(logging::test_logger())
                 .deterministic_keypairs(VALIDATOR_COUNT)
                 .fresh_ephemeral_store()
-                .mock_execution_layer_with_builder(beacon_url.clone())
+                .mock_execution_layer_with_builder(beacon_url.clone(), config.builder_threshold)
                 .build(),
         );
 
@@ -358,6 +376,28 @@ impl ApiTester {
         tester
     }
 
+    pub async fn new_mev_tester_no_builder_threshold() -> Self {
+        let mut config = ApiTesterConfig {
+            builder_threshold: Some(0),
+            spec: E::default_spec(),
+        };
+        config.spec.altair_fork_epoch = Some(Epoch::new(0));
+        config.spec.bellatrix_fork_epoch = Some(Epoch::new(0));
+        let tester = Self::new_from_config(config)
+            .await
+            .test_post_validator_register_validator()
+            .await;
+        tester
+            .mock_builder
+            .as_ref()
+            .unwrap()
+            .builder
+            .add_operation(Operation::Value(Uint256::from(
+                DEFAULT_BUILDER_PAYLOAD_VALUE_WEI,
+            )));
+        tester
+    }
+
     fn skip_slots(self, count: u64) -> Self {
         for _ in 0..count {
             self.chain
@@ -1372,9 +1412,9 @@ impl ApiTester {
     pub async fn test_get_config_spec(self) -> Self {
         let result = self
             .client
-            .get_config_spec::<ConfigAndPresetBellatrix>()
+            .get_config_spec::<ConfigAndPresetCapella>()
             .await
-            .map(|res| ConfigAndPreset::Bellatrix(res.data))
+            .map(|res| ConfigAndPreset::Capella(res.data))
             .unwrap();
         let expected = ConfigAndPreset::from_chain_spec::<E>(&self.chain.spec, None);
 
@@ -2122,7 +2162,7 @@ impl ApiTester {
         self
     }
 
-    pub async fn test_blinded_block_production<Payload: ExecPayload<E>>(&self) {
+    pub async fn test_blinded_block_production<Payload: AbstractExecPayload<E>>(&self) {
         let fork = self.chain.canonical_head.cached_head().head_fork();
         let genesis_validators_root = self.chain.genesis_validators_root;
 
@@ -2182,7 +2222,7 @@ impl ApiTester {
         }
     }
 
-    pub async fn test_blinded_block_production_no_verify_randao<Payload: ExecPayload<E>>(
+    pub async fn test_blinded_block_production_no_verify_randao<Payload: AbstractExecPayload<E>>(
         self,
     ) -> Self {
         for _ in 0..E::slots_per_epoch() {
@@ -2206,7 +2246,9 @@ impl ApiTester {
         self
     }
 
-    pub async fn test_blinded_block_production_verify_randao_invalid<Payload: ExecPayload<E>>(
+    pub async fn test_blinded_block_production_verify_randao_invalid<
+        Payload: AbstractExecPayload<E>,
+    >(
         self,
     ) -> Self {
         let fork = self.chain.canonical_head.cached_head().head_fork();
@@ -2664,7 +2706,7 @@ impl ApiTester {
 
         let (proposer_index, randao_reveal) = self.get_test_randao(slot, epoch).await;
 
-        let payload = self
+        let payload: BlindedPayload<E> = self
             .client
             .get_validator_blinded_blocks::<E, BlindedPayload<E>>(slot, &randao_reveal, None)
             .await
@@ -2673,14 +2715,11 @@ impl ApiTester {
             .body()
             .execution_payload()
             .unwrap()
-            .clone();
+            .into();
 
         let expected_fee_recipient = Address::from_low_u64_be(proposer_index as u64);
-        assert_eq!(
-            payload.execution_payload_header.fee_recipient,
-            expected_fee_recipient
-        );
-        assert_eq!(payload.execution_payload_header.gas_limit, 11_111_111);
+        assert_eq!(payload.fee_recipient(), expected_fee_recipient);
+        assert_eq!(payload.gas_limit(), 11_111_111);
 
         // If this cache is empty, it indicates fallback was not used, so the payload came from the
         // mock builder.
@@ -2707,7 +2746,7 @@ impl ApiTester {
 
         let (proposer_index, randao_reveal) = self.get_test_randao(slot, epoch).await;
 
-        let payload = self
+        let payload: BlindedPayload<E> = self
             .client
             .get_validator_blinded_blocks::<E, BlindedPayload<E>>(slot, &randao_reveal, None)
             .await
@@ -2716,14 +2755,11 @@ impl ApiTester {
             .body()
             .execution_payload()
             .unwrap()
-            .clone();
+            .into();
 
         let expected_fee_recipient = Address::from_low_u64_be(proposer_index as u64);
-        assert_eq!(
-            payload.execution_payload_header.fee_recipient,
-            expected_fee_recipient
-        );
-        assert_eq!(payload.execution_payload_header.gas_limit, 30_000_000);
+        assert_eq!(payload.fee_recipient(), expected_fee_recipient);
+        assert_eq!(payload.gas_limit(), 30_000_000);
 
         // This cache should not be populated because fallback should not have been used.
         assert!(self
@@ -2753,7 +2789,7 @@ impl ApiTester {
 
         let (_, randao_reveal) = self.get_test_randao(slot, epoch).await;
 
-        let payload = self
+        let payload: BlindedPayload<E> = self
             .client
             .get_validator_blinded_blocks::<E, BlindedPayload<E>>(slot, &randao_reveal, None)
             .await
@@ -2762,12 +2798,9 @@ impl ApiTester {
             .body()
             .execution_payload()
             .unwrap()
-            .clone();
+            .into();
 
-        assert_eq!(
-            payload.execution_payload_header.fee_recipient,
-            test_fee_recipient
-        );
+        assert_eq!(payload.fee_recipient(), test_fee_recipient);
 
         // This cache should not be populated because fallback should not have been used.
         assert!(self
@@ -2801,11 +2834,11 @@ impl ApiTester {
             .beacon_state
             .latest_execution_payload_header()
             .unwrap()
-            .block_hash;
+            .block_hash();
 
         let (_, randao_reveal) = self.get_test_randao(slot, epoch).await;
 
-        let payload = self
+        let payload: BlindedPayload<E> = self
             .client
             .get_validator_blinded_blocks::<E, BlindedPayload<E>>(slot, &randao_reveal, None)
             .await
@@ -2814,12 +2847,9 @@ impl ApiTester {
             .body()
             .execution_payload()
             .unwrap()
-            .clone();
+            .into();
 
-        assert_eq!(
-            payload.execution_payload_header.parent_hash,
-            expected_parent_hash
-        );
+        assert_eq!(payload.parent_hash(), expected_parent_hash);
 
         // If this cache is populated, it indicates fallback to the local EE was correctly used.
         assert!(self
@@ -2856,7 +2886,7 @@ impl ApiTester {
 
         let (_, randao_reveal) = self.get_test_randao(slot, epoch).await;
 
-        let payload = self
+        let payload: BlindedPayload<E> = self
             .client
             .get_validator_blinded_blocks::<E, BlindedPayload<E>>(slot, &randao_reveal, None)
             .await
@@ -2865,12 +2895,9 @@ impl ApiTester {
             .body()
             .execution_payload()
             .unwrap()
-            .clone();
+            .into();
 
-        assert_eq!(
-            payload.execution_payload_header.prev_randao,
-            expected_prev_randao
-        );
+        assert_eq!(payload.prev_randao(), expected_prev_randao);
 
         // If this cache is populated, it indicates fallback to the local EE was correctly used.
         assert!(self
@@ -2901,12 +2928,12 @@ impl ApiTester {
             .beacon_state
             .latest_execution_payload_header()
             .unwrap()
-            .block_number
+            .block_number()
             + 1;
 
         let (_, randao_reveal) = self.get_test_randao(slot, epoch).await;
 
-        let payload = self
+        let payload: BlindedPayload<E> = self
             .client
             .get_validator_blinded_blocks::<E, BlindedPayload<E>>(slot, &randao_reveal, None)
             .await
@@ -2915,12 +2942,9 @@ impl ApiTester {
             .body()
             .execution_payload()
             .unwrap()
-            .clone();
+            .into();
 
-        assert_eq!(
-            payload.execution_payload_header.block_number,
-            expected_block_number
-        );
+        assert_eq!(payload.block_number(), expected_block_number);
 
         // If this cache is populated, it indicates fallback to the local EE was correctly used.
         assert!(self
@@ -2951,11 +2975,11 @@ impl ApiTester {
             .beacon_state
             .latest_execution_payload_header()
             .unwrap()
-            .timestamp;
+            .timestamp();
 
         let (_, randao_reveal) = self.get_test_randao(slot, epoch).await;
 
-        let payload = self
+        let payload: BlindedPayload<E> = self
             .client
             .get_validator_blinded_blocks::<E, BlindedPayload<E>>(slot, &randao_reveal, None)
             .await
@@ -2964,9 +2988,9 @@ impl ApiTester {
             .body()
             .execution_payload()
             .unwrap()
-            .clone();
+            .into();
 
-        assert!(payload.execution_payload_header.timestamp > min_expected_timestamp);
+        assert!(payload.timestamp() > min_expected_timestamp);
 
         // If this cache is populated, it indicates fallback to the local EE was correctly used.
         assert!(self
@@ -2991,7 +3015,7 @@ impl ApiTester {
 
         let (_, randao_reveal) = self.get_test_randao(slot, epoch).await;
 
-        let payload = self
+        let payload: BlindedPayload<E> = self
             .client
             .get_validator_blinded_blocks::<E, BlindedPayload<E>>(slot, &randao_reveal, None)
             .await
@@ -3000,7 +3024,7 @@ impl ApiTester {
             .body()
             .execution_payload()
             .unwrap()
-            .clone();
+            .into();
 
         // If this cache is populated, it indicates fallback to the local EE was correctly used.
         assert!(self
@@ -3028,7 +3052,7 @@ impl ApiTester {
 
         let (_, randao_reveal) = self.get_test_randao(slot, epoch).await;
 
-        let payload = self
+        let payload: BlindedPayload<E> = self
             .client
             .get_validator_blinded_blocks::<E, BlindedPayload<E>>(slot, &randao_reveal, None)
             .await
@@ -3037,7 +3061,7 @@ impl ApiTester {
             .body()
             .execution_payload()
             .unwrap()
-            .clone();
+            .into();
 
         // If this cache is populated, it indicates fallback to the local EE was correctly used.
         assert!(self
@@ -3071,7 +3095,7 @@ impl ApiTester {
             .get_test_randao(next_slot, next_slot.epoch(E::slots_per_epoch()))
             .await;
 
-        let payload = self
+        let payload: BlindedPayload<E> = self
             .client
             .get_validator_blinded_blocks::<E, BlindedPayload<E>>(next_slot, &randao_reveal, None)
             .await
@@ -3080,7 +3104,7 @@ impl ApiTester {
             .body()
             .execution_payload()
             .unwrap()
-            .clone();
+            .into();
 
         // This cache should not be populated because fallback should not have been used.
         assert!(self
@@ -3100,7 +3124,7 @@ impl ApiTester {
             .get_test_randao(next_slot, next_slot.epoch(E::slots_per_epoch()))
             .await;
 
-        let payload = self
+        let payload: BlindedPayload<E> = self
             .client
             .get_validator_blinded_blocks::<E, BlindedPayload<E>>(next_slot, &randao_reveal, None)
             .await
@@ -3109,7 +3133,7 @@ impl ApiTester {
             .body()
             .execution_payload()
             .unwrap()
-            .clone();
+            .into();
 
         // If this cache is populated, it indicates fallback to the local EE was correctly used.
         assert!(self
@@ -3149,7 +3173,7 @@ impl ApiTester {
             .get_test_randao(next_slot, next_slot.epoch(E::slots_per_epoch()))
             .await;
 
-        let payload = self
+        let payload: BlindedPayload<E> = self
             .client
             .get_validator_blinded_blocks::<E, BlindedPayload<E>>(next_slot, &randao_reveal, None)
             .await
@@ -3158,7 +3182,7 @@ impl ApiTester {
             .body()
             .execution_payload()
             .unwrap()
-            .clone();
+            .into();
 
         // If this cache is populated, it indicates fallback to the local EE was correctly used.
         assert!(self
@@ -3188,7 +3212,7 @@ impl ApiTester {
             .get_test_randao(next_slot, next_slot.epoch(E::slots_per_epoch()))
             .await;
 
-        let payload = self
+        let payload: BlindedPayload<E> = self
             .client
             .get_validator_blinded_blocks::<E, BlindedPayload<E>>(next_slot, &randao_reveal, None)
             .await
@@ -3197,7 +3221,7 @@ impl ApiTester {
             .body()
             .execution_payload()
             .unwrap()
-            .clone();
+            .into();
 
         // This cache should not be populated because fallback should not have been used.
         assert!(self
@@ -3231,7 +3255,7 @@ impl ApiTester {
 
         let (proposer_index, randao_reveal) = self.get_test_randao(slot, epoch).await;
 
-        let payload = self
+        let payload: BlindedPayload<E> = self
             .client
             .get_validator_blinded_blocks::<E, BlindedPayload<E>>(slot, &randao_reveal, None)
             .await
@@ -3240,13 +3264,10 @@ impl ApiTester {
             .body()
             .execution_payload()
             .unwrap()
-            .clone();
+            .into();
 
         let expected_fee_recipient = Address::from_low_u64_be(proposer_index as u64);
-        assert_eq!(
-            payload.execution_payload_header.fee_recipient,
-            expected_fee_recipient
-        );
+        assert_eq!(payload.fee_recipient(), expected_fee_recipient);
 
         // If this cache is populated, it indicates fallback to the local EE was correctly used.
         assert!(self
@@ -3275,7 +3296,7 @@ impl ApiTester {
 
         let (_, randao_reveal) = self.get_test_randao(slot, epoch).await;
 
-        let payload = self
+        let payload: BlindedPayload<E> = self
             .client
             .get_validator_blinded_blocks::<E, BlindedPayload<E>>(slot, &randao_reveal, None)
             .await
@@ -3284,7 +3305,7 @@ impl ApiTester {
             .body()
             .execution_payload()
             .unwrap()
-            .clone();
+            .into();
 
         // If this cache is populated, it indicates fallback to the local EE was correctly used.
         assert!(self
@@ -3297,6 +3318,209 @@ impl ApiTester {
         self
     }
 
+    pub async fn test_builder_payload_chosen_when_more_profitable(self) -> Self {
+        // Mutate value.
+        self.mock_builder
+            .as_ref()
+            .unwrap()
+            .builder
+            .add_operation(Operation::Value(Uint256::from(
+                DEFAULT_MOCK_EL_PAYLOAD_VALUE_WEI + 1,
+            )));
+
+        let slot = self.chain.slot().unwrap();
+        let epoch = self.chain.epoch().unwrap();
+
+        let (_, randao_reveal) = self.get_test_randao(slot, epoch).await;
+
+        let payload: BlindedPayload<E> = self
+            .client
+            .get_validator_blinded_blocks::<E, BlindedPayload<E>>(slot, &randao_reveal, None)
+            .await
+            .unwrap()
+            .data
+            .body()
+            .execution_payload()
+            .unwrap()
+            .into();
+
+        // The builder's payload should've been chosen, so this cache should not be populated
+        assert!(self
+            .chain
+            .execution_layer
+            .as_ref()
+            .unwrap()
+            .get_payload_by_root(&payload.tree_hash_root())
+            .is_none());
+        self
+    }
+
+    pub async fn test_local_payload_chosen_when_equally_profitable(self) -> Self {
+        // Mutate value.
+        self.mock_builder
+            .as_ref()
+            .unwrap()
+            .builder
+            .add_operation(Operation::Value(Uint256::from(
+                DEFAULT_MOCK_EL_PAYLOAD_VALUE_WEI,
+            )));
+
+        let slot = self.chain.slot().unwrap();
+        let epoch = self.chain.epoch().unwrap();
+
+        let (_, randao_reveal) = self.get_test_randao(slot, epoch).await;
+
+        let payload: BlindedPayload<E> = self
+            .client
+            .get_validator_blinded_blocks::<E, BlindedPayload<E>>(slot, &randao_reveal, None)
+            .await
+            .unwrap()
+            .data
+            .body()
+            .execution_payload()
+            .unwrap()
+            .into();
+
+        // The local payload should've been chosen, so this cache should be populated
+        assert!(self
+            .chain
+            .execution_layer
+            .as_ref()
+            .unwrap()
+            .get_payload_by_root(&payload.tree_hash_root())
+            .is_some());
+        self
+    }
+
+    pub async fn test_local_payload_chosen_when_more_profitable(self) -> Self {
+        // Mutate value.
+        self.mock_builder
+            .as_ref()
+            .unwrap()
+            .builder
+            .add_operation(Operation::Value(Uint256::from(
+                DEFAULT_MOCK_EL_PAYLOAD_VALUE_WEI - 1,
+            )));
+
+        let slot = self.chain.slot().unwrap();
+        let epoch = self.chain.epoch().unwrap();
+
+        let (_, randao_reveal) = self.get_test_randao(slot, epoch).await;
+
+        let payload: BlindedPayload<E> = self
+            .client
+            .get_validator_blinded_blocks::<E, BlindedPayload<E>>(slot, &randao_reveal, None)
+            .await
+            .unwrap()
+            .data
+            .body()
+            .execution_payload()
+            .unwrap()
+            .into();
+
+        // The local payload should've been chosen, so this cache should be populated
+        assert!(self
+            .chain
+            .execution_layer
+            .as_ref()
+            .unwrap()
+            .get_payload_by_root(&payload.tree_hash_root())
+            .is_some());
+        self
+    }
+
+    pub async fn test_builder_works_post_capella(self) -> Self {
+        // Ensure builder payload is chosen
+        self.mock_builder
+            .as_ref()
+            .unwrap()
+            .builder
+            .add_operation(Operation::Value(Uint256::from(
+                DEFAULT_MOCK_EL_PAYLOAD_VALUE_WEI + 1,
+            )));
+
+        let slot = self.chain.slot().unwrap();
+        let propose_state = self
+            .harness
+            .chain
+            .state_at_slot(slot, StateSkipConfig::WithoutStateRoots)
+            .unwrap();
+        let withdrawals = get_expected_withdrawals(&propose_state, &self.chain.spec).unwrap();
+        let withdrawals_root = withdrawals.tree_hash_root();
+        // Set withdrawals root for builder
+        self.mock_builder
+            .as_ref()
+            .unwrap()
+            .builder
+            .add_operation(Operation::WithdrawalsRoot(withdrawals_root));
+
+        let epoch = self.chain.epoch().unwrap();
+        let (_, randao_reveal) = self.get_test_randao(slot, epoch).await;
+
+        let payload: BlindedPayload<E> = self
+            .client
+            .get_validator_blinded_blocks::<E, BlindedPayload<E>>(slot, &randao_reveal, None)
+            .await
+            .unwrap()
+            .data
+            .body()
+            .execution_payload()
+            .unwrap()
+            .into();
+
+        // The builder's payload should've been chosen, so this cache should not be populated
+        assert!(self
+            .chain
+            .execution_layer
+            .as_ref()
+            .unwrap()
+            .get_payload_by_root(&payload.tree_hash_root())
+            .is_none());
+        self
+    }
+
+    pub async fn test_lighthouse_rejects_invalid_withdrawals_root(self) -> Self {
+        // Ensure builder payload *would be* chosen
+        self.mock_builder
+            .as_ref()
+            .unwrap()
+            .builder
+            .add_operation(Operation::Value(Uint256::from(
+                DEFAULT_MOCK_EL_PAYLOAD_VALUE_WEI + 1,
+            )));
+        // Set withdrawals root to something invalid
+        self.mock_builder
+            .as_ref()
+            .unwrap()
+            .builder
+            .add_operation(Operation::WithdrawalsRoot(Hash256::repeat_byte(0x42)));
+
+        let slot = self.chain.slot().unwrap();
+        let epoch = self.chain.epoch().unwrap();
+        let (_, randao_reveal) = self.get_test_randao(slot, epoch).await;
+
+        let payload: BlindedPayload<E> = self
+            .client
+            .get_validator_blinded_blocks::<E, BlindedPayload<E>>(slot, &randao_reveal, None)
+            .await
+            .unwrap()
+            .data
+            .body()
+            .execution_payload()
+            .unwrap()
+            .into();
+
+        // The local payload should've been chosen because the builder's was invalid
+        assert!(self
+            .chain
+            .execution_layer
+            .as_ref()
+            .unwrap()
+            .get_payload_by_root(&payload.tree_hash_root())
+            .is_some());
+        self
+    }
+
     #[cfg(target_os = "linux")]
     pub async fn test_get_lighthouse_health(self) -> Self {
         self.client.get_lighthouse_health().await.unwrap();
@@ -3766,9 +3990,9 @@ async fn get_events() {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn get_events_altair() {
-    let mut spec = E::default_spec();
-    spec.altair_fork_epoch = Some(Epoch::new(0));
-    ApiTester::new_from_spec(spec)
+    let mut config = ApiTesterConfig::default();
+    config.spec.altair_fork_epoch = Some(Epoch::new(0));
+    ApiTester::new_from_config(config)
         .await
         .test_get_events_altair()
         .await;
@@ -4281,6 +4505,38 @@ async fn builder_inadequate_builder_threshold() {
         .await;
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn builder_payload_chosen_by_profit() {
+    ApiTester::new_mev_tester_no_builder_threshold()
+        .await
+        .test_builder_payload_chosen_when_more_profitable()
+        .await
+        .test_local_payload_chosen_when_equally_profitable()
+        .await
+        .test_local_payload_chosen_when_more_profitable()
+        .await;
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn builder_works_post_capella() {
+    let mut config = ApiTesterConfig {
+        builder_threshold: Some(0),
+        spec: E::default_spec(),
+    };
+    config.spec.altair_fork_epoch = Some(Epoch::new(0));
+    config.spec.bellatrix_fork_epoch = Some(Epoch::new(0));
+    config.spec.capella_fork_epoch = Some(Epoch::new(0));
+
+    ApiTester::new_from_config(config)
+        .await
+        .test_post_validator_register_validator()
+        .await
+        .test_builder_works_post_capella()
+        .await
+        .test_lighthouse_rejects_invalid_withdrawals_root()
+        .await;
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn lighthouse_endpoints() {
     ApiTester::new()