Implement expected withdrawals endpoint (#4390)

## Issue Addressed

[#4029](https://github.com/sigp/lighthouse/issues/4029)

## Proposed Changes

implement expected_withdrawals HTTP API per the spec 

https://github.com/ethereum/beacon-APIs/pull/304

## Additional Info

beacon_node/http_api/src/builder_states.rs

@@ -0,0 +1,72 @@
+use crate::StateId;
+use beacon_chain::{BeaconChain, BeaconChainTypes};
+use safe_arith::SafeArith;
+use state_processing::per_block_processing::get_expected_withdrawals;
+use state_processing::state_advance::partial_state_advance;
+use std::sync::Arc;
+use types::{BeaconState, EthSpec, ForkName, Slot, Withdrawals};
+
+const MAX_EPOCH_LOOKAHEAD: u64 = 2;
+
+/// Get the withdrawals computed from the specified state, that will be included in the block
+/// that gets built on the specified state.
+pub fn get_next_withdrawals<T: BeaconChainTypes>(
+    chain: &Arc<BeaconChain<T>>,
+    mut state: BeaconState<T::EthSpec>,
+    state_id: StateId,
+    proposal_slot: Slot,
+) -> Result<Withdrawals<T::EthSpec>, warp::Rejection> {
+    get_next_withdrawals_sanity_checks(chain, &state, proposal_slot)?;
+
+    // advance the state to the epoch of the proposal slot.
+    let proposal_epoch = proposal_slot.epoch(T::EthSpec::slots_per_epoch());
+    let (state_root, _, _) = state_id.root(chain)?;
+    if proposal_epoch != state.current_epoch() {
+        if let Err(e) =
+            partial_state_advance(&mut state, Some(state_root), proposal_slot, &chain.spec)
+        {
+            return Err(warp_utils::reject::custom_server_error(format!(
+                "failed to advance to the epoch of the proposal slot: {:?}",
+                e
+            )));
+        }
+    }
+
+    match get_expected_withdrawals(&state, &chain.spec) {
+        Ok(withdrawals) => Ok(withdrawals),
+        Err(e) => Err(warp_utils::reject::custom_server_error(format!(
+            "failed to get expected withdrawal: {:?}",
+            e
+        ))),
+    }
+}
+
+fn get_next_withdrawals_sanity_checks<T: BeaconChainTypes>(
+    chain: &BeaconChain<T>,
+    state: &BeaconState<T::EthSpec>,
+    proposal_slot: Slot,
+) -> Result<(), warp::Rejection> {
+    if proposal_slot <= state.slot() {
+        return Err(warp_utils::reject::custom_bad_request(
+            "proposal slot must be greater than the pre-state slot".to_string(),
+        ));
+    }
+
+    let fork = chain.spec.fork_name_at_slot::<T::EthSpec>(proposal_slot);
+    if let ForkName::Base | ForkName::Altair | ForkName::Merge = fork {
+        return Err(warp_utils::reject::custom_bad_request(
+            "the specified state is a pre-capella state.".to_string(),
+        ));
+    }
+
+    let look_ahead_limit = MAX_EPOCH_LOOKAHEAD
+        .safe_mul(T::EthSpec::slots_per_epoch())
+        .map_err(warp_utils::reject::arith_error)?;
+    if proposal_slot >= state.slot() + look_ahead_limit {
+        return Err(warp_utils::reject::custom_bad_request(format!(
+            "proposal slot is greater than or equal to the look ahead limit: {look_ahead_limit}"
+        )));
+    }
+
+    Ok(())
+}

beacon_node/http_api/src/lib.rs

@@ -10,6 +10,7 @@ mod attester_duties;
 mod block_id;
 mod block_packing_efficiency;
 mod block_rewards;
+mod builder_states;
 mod database;
 mod metrics;
 mod proposer_duties;
@@ -32,6 +33,7 @@ use beacon_chain::{
 };
 use beacon_processor::BeaconProcessorSend;
 pub use block_id::BlockId;
+use builder_states::get_next_withdrawals;
 use bytes::Bytes;
 use directory::DEFAULT_ROOT_DIR;
 use eth2::types::{
@@ -2291,6 +2293,60 @@ pub fn serve<T: BeaconChainTypes>(
             },
         );
 
+    /*
+     * builder/states
+     */
+
+    let builder_states_path = eth_v1
+        .and(warp::path("builder"))
+        .and(warp::path("states"))
+        .and(chain_filter.clone());
+
+    // GET builder/states/{state_id}/expected_withdrawals
+    let get_expected_withdrawals = builder_states_path
+        .clone()
+        .and(task_spawner_filter.clone())
+        .and(warp::path::param::<StateId>())
+        .and(warp::path("expected_withdrawals"))
+        .and(warp::query::<api_types::ExpectedWithdrawalsQuery>())
+        .and(warp::path::end())
+        .and(warp::header::optional::<api_types::Accept>("accept"))
+        .then(
+            |chain: Arc<BeaconChain<T>>,
+             task_spawner: TaskSpawner<T::EthSpec>,
+             state_id: StateId,
+             query: api_types::ExpectedWithdrawalsQuery,
+             accept_header: Option<api_types::Accept>| {
+                task_spawner.blocking_response_task(Priority::P1, move || {
+                    let (state, execution_optimistic, finalized) = state_id.state(&chain)?;
+                    let proposal_slot = query.proposal_slot.unwrap_or(state.slot() + 1);
+                    let withdrawals =
+                        get_next_withdrawals::<T>(&chain, state, state_id, proposal_slot)?;
+
+                    match accept_header {
+                        Some(api_types::Accept::Ssz) => Response::builder()
+                            .status(200)
+                            .header("Content-Type", "application/octet-stream")
+                            .body(withdrawals.as_ssz_bytes().into())
+                            .map_err(|e| {
+                                warp_utils::reject::custom_server_error(format!(
+                                    "failed to create response: {}",
+                                    e
+                                ))
+                            }),
+                        _ => Ok(warp::reply::json(
+                            &api_types::ExecutionOptimisticFinalizedResponse {
+                                data: withdrawals,
+                                execution_optimistic: Some(execution_optimistic),
+                                finalized: Some(finalized),
+                            },
+                        )
+                        .into_response()),
+                    }
+                })
+            },
+        );
+
     /*
      * beacon/rewards
      */
@@ -4503,6 +4559,7 @@ pub fn serve<T: BeaconChainTypes>(
                 .uor(get_lighthouse_block_packing_efficiency)
                 .uor(get_lighthouse_merge_readiness)
                 .uor(get_events)
+                .uor(get_expected_withdrawals)
                 .uor(lighthouse_log_events.boxed())
                 .recover(warp_utils::reject::handle_rejection),
         )

beacon_node/http_api/tests/tests.rs

@@ -28,6 +28,7 @@ use sensitive_url::SensitiveUrl;
 use slot_clock::SlotClock;
 use state_processing::per_block_processing::get_expected_withdrawals;
 use state_processing::per_slot_processing;
+use state_processing::state_advance::partial_state_advance;
 use std::convert::TryInto;
 use std::sync::Arc;
 use tokio::time::Duration;
@@ -4341,6 +4342,72 @@ impl ApiTester {
         self
     }
 
+    pub async fn test_get_expected_withdrawals_invalid_state(self) -> Self {
+        let state_id = CoreStateId::Root(Hash256::zero());
+
+        let result = self.client.get_expected_withdrawals(&state_id).await;
+
+        match result {
+            Err(e) => {
+                assert_eq!(e.status().unwrap(), 404);
+            }
+            _ => panic!("query did not fail correctly"),
+        }
+
+        self
+    }
+
+    pub async fn test_get_expected_withdrawals_capella(self) -> Self {
+        let slot = self.chain.slot().unwrap();
+        let state_id = CoreStateId::Slot(slot);
+
+        // calculate the expected withdrawals
+        let (mut state, _, _) = StateId(state_id).state(&self.chain).unwrap();
+        let proposal_slot = state.slot() + 1;
+        let proposal_epoch = proposal_slot.epoch(E::slots_per_epoch());
+        let (state_root, _, _) = StateId(state_id).root(&self.chain).unwrap();
+        if proposal_epoch != state.current_epoch() {
+            let _ = partial_state_advance(
+                &mut state,
+                Some(state_root),
+                proposal_slot,
+                &self.chain.spec,
+            );
+        }
+        let expected_withdrawals = get_expected_withdrawals(&state, &self.chain.spec).unwrap();
+
+        // fetch expected withdrawals from the client
+        let result = self.client.get_expected_withdrawals(&state_id).await;
+        match result {
+            Ok(withdrawal_response) => {
+                assert_eq!(withdrawal_response.execution_optimistic, Some(false));
+                assert_eq!(withdrawal_response.finalized, Some(false));
+                assert_eq!(withdrawal_response.data, expected_withdrawals.to_vec());
+            }
+            Err(e) => {
+                println!("{:?}", e);
+                panic!("query failed incorrectly");
+            }
+        }
+
+        self
+    }
+
+    pub async fn test_get_expected_withdrawals_pre_capella(self) -> Self {
+        let state_id = CoreStateId::Head;
+
+        let result = self.client.get_expected_withdrawals(&state_id).await;
+
+        match result {
+            Err(e) => {
+                assert_eq!(e.status().unwrap(), 400);
+            }
+            _ => panic!("query did not fail correctly"),
+        }
+
+        self
+    }
+
     pub async fn test_get_events_altair(self) -> Self {
         let topics = vec![EventTopic::ContributionAndProof];
         let mut events_future = self
@@ -5123,3 +5190,37 @@ async fn optimistic_responses() {
         .test_check_optimistic_responses()
         .await;
 }
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn expected_withdrawals_invalid_pre_capella() {
+    let mut config = ApiTesterConfig::default();
+    config.spec.altair_fork_epoch = Some(Epoch::new(0));
+    ApiTester::new_from_config(config)
+        .await
+        .test_get_expected_withdrawals_pre_capella()
+        .await;
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn expected_withdrawals_invalid_state() {
+    let mut config = ApiTesterConfig::default();
+    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_get_expected_withdrawals_invalid_state()
+        .await;
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn expected_withdrawals_valid_capella() {
+    let mut config = ApiTesterConfig::default();
+    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_get_expected_withdrawals_capella()
+        .await;
+}

common/eth2/src/lib.rs

@@ -1261,6 +1261,23 @@ impl BeaconNodeHttpClient {
         Ok(())
     }
 
+    // GET builder/states/{state_id}/expected_withdrawals
+    pub async fn get_expected_withdrawals(
+        &self,
+        state_id: &StateId,
+    ) -> Result<ExecutionOptimisticFinalizedResponse<Vec<Withdrawal>>, Error> {
+        let mut path = self.eth_path(V1)?;
+
+        path.path_segments_mut()
+            .map_err(|()| Error::InvalidUrl(self.server.clone()))?
+            .push("builder")
+            .push("states")
+            .push(&state_id.to_string())
+            .push("expected_withdrawals");
+
+        self.get(path).await
+    }
+
     /// `POST validator/contribution_and_proofs`
     pub async fn post_validator_contribution_and_proofs<T: EthSpec>(
         &self,

common/eth2/src/types.rs

@@ -581,6 +581,11 @@ pub struct SyncingData {
     pub sync_distance: Slot,
 }
 
+#[derive(Serialize, Deserialize)]
+pub struct ExpectedWithdrawalsQuery {
+    pub proposal_slot: Option<Slot>,
+}
+
 #[derive(Clone, PartialEq, Debug, Deserialize)]
 #[serde(try_from = "String", bound = "T: FromStr")]
 pub struct QueryVec<T: FromStr> {