Add API version headers and map_fork_name! (#2745)

## Proposed Changes

* Add the `Eth-Consensus-Version` header to the HTTP API for the block and state endpoints. This is part of the v2.1.0 API that was recently released: ethereum/beacon-APIs#170
* Add tests for the above. I refactored the `eth2` crate's helper functions to make this more straight-forward, and introduced some new mixin traits that I think greatly improve readability and flexibility.
* Add a new `map_with_fork!` macro which is useful for decoding a superstruct type without naming all its variants. It is now used for SSZ-decoding `BeaconBlock` and `BeaconState`, and for JSON-decoding `SignedBeaconBlock` in the API.

## Additional Info

The `map_with_fork!` changes will conflict with the Merge changes, but when resolving the conflict the changes from this branch should be preferred (it is no longer necessary to enumerate every fork). The merge fork _will_  need to be added to `map_fork_name_with`.

beacon_node/http_api/src/lib.rs

@@ -47,7 +47,10 @@ use types::{
     SignedContributionAndProof, SignedVoluntaryExit, Slot, SyncCommitteeMessage,
     SyncContributionData,
 };
-use version::{fork_versioned_response, unsupported_version_rejection, V1};
+use version::{
+    add_consensus_version_header, fork_versioned_response, inconsistent_fork_rejection,
+    unsupported_version_rejection, V1,
+};
 use warp::http::StatusCode;
 use warp::sse::Event;
 use warp::Reply;
@@ -1003,6 +1006,9 @@ pub fn serve<T: BeaconChainTypes>(
              accept_header: Option<api_types::Accept>| {
                 blocking_task(move || {
                     let block = block_id.block(&chain)?;
+                    let fork_name = block
+                        .fork_name(&chain.spec)
+                        .map_err(inconsistent_fork_rejection)?;
                     match accept_header {
                         Some(api_types::Accept::Ssz) => Response::builder()
                             .status(200)
@@ -1014,12 +1020,10 @@ pub fn serve<T: BeaconChainTypes>(
                                     e
                                 ))
                             }),
-                        _ => {
-                            let fork_name = block.fork_name(&chain.spec).ok();
-                            fork_versioned_response(endpoint_version, fork_name, block)
-                                .map(|res| warp::reply::json(&res).into_response())
-                        }
+                        _ => fork_versioned_response(endpoint_version, fork_name, block)
+                            .map(|res| warp::reply::json(&res).into_response()),
                     }
+                    .map(|resp| add_consensus_version_header(resp, fork_name))
                 })
             },
         );
@@ -1459,10 +1463,14 @@ pub fn serve<T: BeaconChainTypes>(
                 blocking_task(move || match accept_header {
                     Some(api_types::Accept::Ssz) => {
                         let state = state_id.state(&chain)?;
+                        let fork_name = state
+                            .fork_name(&chain.spec)
+                            .map_err(inconsistent_fork_rejection)?;
                         Response::builder()
                             .status(200)
                             .header("Content-Type", "application/octet-stream")
                             .body(state.as_ssz_bytes().into())
+                            .map(|resp| add_consensus_version_header(resp, fork_name))
                             .map_err(|e| {
                                 warp_utils::reject::custom_server_error(format!(
                                     "failed to create response: {}",
@@ -1471,9 +1479,14 @@ pub fn serve<T: BeaconChainTypes>(
                             })
                     }
                     _ => state_id.map_state(&chain, |state| {
-                        let fork_name = state.fork_name(&chain.spec).ok();
+                        let fork_name = state
+                            .fork_name(&chain.spec)
+                            .map_err(inconsistent_fork_rejection)?;
                         let res = fork_versioned_response(endpoint_version, fork_name, &state)?;
-                        Ok(warp::reply::json(&res).into_response())
+                        Ok(add_consensus_version_header(
+                            warp::reply::json(&res).into_response(),
+                            fork_name,
+                        ))
                     }),
                 })
             },
@@ -1821,7 +1834,10 @@ pub fn serve<T: BeaconChainTypes>(
                     let (block, _) = chain
                         .produce_block(randao_reveal, slot, query.graffiti.map(Into::into))
                         .map_err(warp_utils::reject::block_production_error)?;
-                    let fork_name = block.to_ref().fork_name(&chain.spec).ok();
+                    let fork_name = block
+                        .to_ref()
+                        .fork_name(&chain.spec)
+                        .map_err(inconsistent_fork_rejection)?;
                     fork_versioned_response(endpoint_version, fork_name, block)
                 })
             },

beacon_node/http_api/src/version.rs

@@ -1,19 +1,21 @@
 use crate::api_types::{EndpointVersion, ForkVersionedResponse};
+use eth2::CONSENSUS_VERSION_HEADER;
 use serde::Serialize;
-use types::ForkName;
+use types::{ForkName, InconsistentFork};
+use warp::reply::{self, Reply, WithHeader};
 
 pub const V1: EndpointVersion = EndpointVersion(1);
 pub const V2: EndpointVersion = EndpointVersion(2);
 
 pub fn fork_versioned_response<T: Serialize>(
     endpoint_version: EndpointVersion,
-    fork_name: Option<ForkName>,
+    fork_name: ForkName,
     data: T,
 ) -> Result<ForkVersionedResponse<T>, warp::reject::Rejection> {
     let fork_name = if endpoint_version == V1 {
         None
     } else if endpoint_version == V2 {
-        fork_name
+        Some(fork_name)
     } else {
         return Err(unsupported_version_rejection(endpoint_version));
     };
@@ -23,6 +25,15 @@ pub fn fork_versioned_response<T: Serialize>(
     })
 }
 
+/// Add the `Eth-Consensus-Version` header to a response.
+pub fn add_consensus_version_header<T: Reply>(reply: T, fork_name: ForkName) -> WithHeader<T> {
+    reply::with_header(reply, CONSENSUS_VERSION_HEADER, fork_name.to_string())
+}
+
+pub fn inconsistent_fork_rejection(error: InconsistentFork) -> warp::reject::Rejection {
+    warp_utils::reject::custom_server_error(format!("wrong fork: {:?}", error))
+}
+
 pub fn unsupported_version_rejection(version: EndpointVersion) -> warp::reject::Rejection {
     warp_utils::reject::custom_bad_request(format!("Unsupported endpoint version: {}", version))
 }

beacon_node/http_api/tests/tests.rs

@@ -5,9 +5,12 @@ use beacon_chain::{
     BeaconChain, StateSkipConfig, WhenSlotSkipped, MAXIMUM_GOSSIP_CLOCK_DISPARITY,
 };
 use environment::null_logger;
-use eth2::Error;
-use eth2::StatusCode;
-use eth2::{types::*, BeaconNodeHttpClient, Timeouts};
+use eth2::{
+    mixin::{RequestAccept, ResponseForkName, ResponseOptional},
+    reqwest::RequestBuilder,
+    types::*,
+    BeaconNodeHttpClient, Error, StatusCode, Timeouts,
+};
 use futures::stream::{Stream, StreamExt};
 use futures::FutureExt;
 use lighthouse_network::{Enr, EnrExt, PeerId};
@@ -952,6 +955,7 @@ impl ApiTester {
                 }
             }
 
+            // Check the JSON endpoint.
             let json_result = self.client.get_beacon_blocks(block_id).await.unwrap();
 
             if let (Some(json), Some(expected)) = (&json_result, &expected) {
@@ -965,6 +969,7 @@ impl ApiTester {
                 assert_eq!(expected, None);
             }
 
+            // Check the SSZ endpoint.
             let ssz_result = self
                 .client
                 .get_beacon_blocks_ssz(block_id, &self.chain.spec)
@@ -981,6 +986,34 @@ impl ApiTester {
                 assert_eq!(v1_result, None);
                 assert_eq!(expected, None);
             }
+
+            // Check that version headers are provided.
+            let url = self.client.get_beacon_blocks_path(block_id).unwrap();
+
+            let builders: Vec<fn(RequestBuilder) -> RequestBuilder> = vec![
+                |b| b,
+                |b| b.accept(Accept::Ssz),
+                |b| b.accept(Accept::Json),
+                |b| b.accept(Accept::Any),
+            ];
+
+            for req_builder in builders {
+                let raw_res = self
+                    .client
+                    .get_response(url.clone(), req_builder)
+                    .await
+                    .optional()
+                    .unwrap();
+                if let (Some(raw_res), Some(expected)) = (&raw_res, &expected) {
+                    assert_eq!(
+                        raw_res.fork_name_from_header().unwrap(),
+                        Some(expected.fork_name(&self.chain.spec).unwrap())
+                    );
+                } else {
+                    assert!(raw_res.is_none());
+                    assert_eq!(expected, None);
+                }
+            }
         }
 
         self
@@ -1440,6 +1473,30 @@ impl ApiTester {
                 assert_eq!(result_v1, None);
                 assert_eq!(expected, None);
             }
+
+            // Check that version headers are provided.
+            let url = self.client.get_debug_beacon_states_path(state_id).unwrap();
+
+            let builders: Vec<fn(RequestBuilder) -> RequestBuilder> =
+                vec![|b| b, |b| b.accept(Accept::Ssz)];
+
+            for req_builder in builders {
+                let raw_res = self
+                    .client
+                    .get_response(url.clone(), req_builder)
+                    .await
+                    .optional()
+                    .unwrap();
+                if let (Some(raw_res), Some(expected)) = (&raw_res, &expected) {
+                    assert_eq!(
+                        raw_res.fork_name_from_header().unwrap(),
+                        Some(expected.fork_name(&self.chain.spec).unwrap())
+                    );
+                } else {
+                    assert!(raw_res.is_none());
+                    assert_eq!(expected, None);
+                }
+            }
         }
 
         self

beacon_node/store/src/partial_beacon_state.rs

@@ -176,16 +176,13 @@ impl<T: EthSpec> PartialBeaconState<T> {
         )?;
 
         let slot = Slot::from_ssz_bytes(slot_bytes)?;
-        let epoch = slot.epoch(T::slots_per_epoch());
-
-        if spec
-            .altair_fork_epoch
-            .map_or(true, |altair_epoch| epoch < altair_epoch)
-        {
-            PartialBeaconStateBase::from_ssz_bytes(bytes).map(Self::Base)
-        } else {
-            PartialBeaconStateAltair::from_ssz_bytes(bytes).map(Self::Altair)
-        }
+        let fork_at_slot = spec.fork_name_at_slot::<T>(slot);
+
+        Ok(map_fork_name!(
+            fork_at_slot,
+            Self,
+            <_>::from_ssz_bytes(bytes)?
+        ))
     }
 
     /// Prepare the partial state for storage in the KV database.