Merge remote-tracking branch 'origin/unstable' into tree-states

beacon_node/http_api/src/lib.rs

@@ -31,8 +31,8 @@ use beacon_chain::{
 pub use block_id::BlockId;
 use directory::DEFAULT_ROOT_DIR;
 use eth2::types::{
-    self as api_types, EndpointVersion, ForkChoice, ForkChoiceNode, SkipRandaoVerification,
-    ValidatorId, ValidatorStatus,
+    self as api_types, BroadcastValidation, EndpointVersion, ForkChoice, ForkChoiceNode,
+    SkipRandaoVerification, ValidatorId, ValidatorStatus,
 };
 use lighthouse_network::{types::SyncState, EnrExt, NetworkGlobals, PeerId, PubsubMessage};
 use lighthouse_version::version_with_platform;
@@ -40,7 +40,9 @@ use logging::SSELoggingComponents;
 use network::{NetworkMessage, NetworkSenders, ValidatorSubscriptionMessage};
 use operation_pool::ReceivedPreCapella;
 use parking_lot::RwLock;
-use publish_blocks::ProvenancedBlock;
+pub use publish_blocks::{
+    publish_blinded_block, publish_block, reconstruct_block, ProvenancedBlock,
+};
 use serde::{Deserialize, Serialize};
 use slog::{crit, debug, error, info, warn, Logger};
 use slot_clock::SlotClock;
@@ -323,6 +325,7 @@ pub fn serve<T: BeaconChainTypes>(
     };
 
     let eth_v1 = single_version(V1);
+    let eth_v2 = single_version(V2);
 
     // Create a `warp` filter that provides access to the network globals.
     let inner_network_globals = ctx.network_globals.clone();
@@ -1240,16 +1243,55 @@ pub fn serve<T: BeaconChainTypes>(
              log: Logger| async move {
                 publish_blocks::publish_block(
                     None,
-                    ProvenancedBlock::Local(block),
+                    ProvenancedBlock::local(block),
                     chain,
                     &network_tx,
                     log,
+                    BroadcastValidation::default(),
                 )
                 .await
                 .map(|()| warp::reply().into_response())
             },
         );
 
+    let post_beacon_blocks_v2 = eth_v2
+        .and(warp::path("beacon"))
+        .and(warp::path("blocks"))
+        .and(warp::query::<api_types::BroadcastValidationQuery>())
+        .and(warp::path::end())
+        .and(warp::body::json())
+        .and(chain_filter.clone())
+        .and(network_tx_filter.clone())
+        .and(log_filter.clone())
+        .then(
+            |validation_level: api_types::BroadcastValidationQuery,
+             block: Arc<SignedBeaconBlock<T::EthSpec>>,
+             chain: Arc<BeaconChain<T>>,
+             network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>,
+             log: Logger| async move {
+                match publish_blocks::publish_block(
+                    None,
+                    ProvenancedBlock::local(block),
+                    chain,
+                    &network_tx,
+                    log,
+                    validation_level.broadcast_validation,
+                )
+                .await
+                {
+                    Ok(()) => warp::reply().into_response(),
+                    Err(e) => match warp_utils::reject::handle_rejection(e).await {
+                        Ok(reply) => reply.into_response(),
+                        Err(_) => warp::reply::with_status(
+                            StatusCode::INTERNAL_SERVER_ERROR,
+                            eth2::StatusCode::INTERNAL_SERVER_ERROR,
+                        )
+                        .into_response(),
+                    },
+                }
+            },
+        );
+
     /*
      * beacon/blocks
      */
@@ -1268,9 +1310,52 @@ pub fn serve<T: BeaconChainTypes>(
              chain: Arc<BeaconChain<T>>,
              network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>,
              log: Logger| async move {
-                publish_blocks::publish_blinded_block(block, chain, &network_tx, log)
-                    .await
-                    .map(|()| warp::reply().into_response())
+                publish_blocks::publish_blinded_block(
+                    block,
+                    chain,
+                    &network_tx,
+                    log,
+                    BroadcastValidation::default(),
+                )
+                .await
+                .map(|()| warp::reply().into_response())
+            },
+        );
+
+    let post_beacon_blinded_blocks_v2 = eth_v2
+        .and(warp::path("beacon"))
+        .and(warp::path("blinded_blocks"))
+        .and(warp::query::<api_types::BroadcastValidationQuery>())
+        .and(warp::path::end())
+        .and(warp::body::json())
+        .and(chain_filter.clone())
+        .and(network_tx_filter.clone())
+        .and(log_filter.clone())
+        .then(
+            |validation_level: api_types::BroadcastValidationQuery,
+             block: SignedBeaconBlock<T::EthSpec, BlindedPayload<_>>,
+             chain: Arc<BeaconChain<T>>,
+             network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>,
+             log: Logger| async move {
+                match publish_blocks::publish_blinded_block(
+                    block,
+                    chain,
+                    &network_tx,
+                    log,
+                    validation_level.broadcast_validation,
+                )
+                .await
+                {
+                    Ok(()) => warp::reply().into_response(),
+                    Err(e) => match warp_utils::reject::handle_rejection(e).await {
+                        Ok(reply) => reply.into_response(),
+                        Err(_) => warp::reply::with_status(
+                            StatusCode::INTERNAL_SERVER_ERROR,
+                            eth2::StatusCode::INTERNAL_SERVER_ERROR,
+                        )
+                        .into_response(),
+                    },
+                }
             },
         );
 
@@ -2351,24 +2436,41 @@ pub fn serve<T: BeaconChainTypes>(
         .and(warp::path("health"))
         .and(warp::path::end())
         .and(network_globals.clone())
-        .and_then(|network_globals: Arc<NetworkGlobals<T::EthSpec>>| {
-            blocking_response_task(move || match *network_globals.sync_state.read() {
-                SyncState::SyncingFinalized { .. }
-                | SyncState::SyncingHead { .. }
-                | SyncState::SyncTransition
-                | SyncState::BackFillSyncing { .. } => Ok(warp::reply::with_status(
-                    warp::reply(),
-                    warp::http::StatusCode::PARTIAL_CONTENT,
-                )),
-                SyncState::Synced => Ok(warp::reply::with_status(
-                    warp::reply(),
-                    warp::http::StatusCode::OK,
-                )),
-                SyncState::Stalled => Err(warp_utils::reject::not_synced(
-                    "sync stalled, beacon chain may not yet be initialized.".to_string(),
-                )),
-            })
-        });
+        .and(chain_filter.clone())
+        .and_then(
+            |network_globals: Arc<NetworkGlobals<T::EthSpec>>, chain: Arc<BeaconChain<T>>| {
+                async move {
+                    let el_offline = if let Some(el) = &chain.execution_layer {
+                        el.is_offline_or_erroring().await
+                    } else {
+                        true
+                    };
+
+                    blocking_response_task(move || {
+                        let is_optimistic = chain
+                            .is_optimistic_or_invalid_head()
+                            .map_err(warp_utils::reject::beacon_chain_error)?;
+
+                        let is_syncing = !network_globals.sync_state.read().is_synced();
+
+                        if el_offline {
+                            Err(warp_utils::reject::not_synced("execution layer is offline".to_string()))
+                        } else if is_syncing || is_optimistic {
+                            Ok(warp::reply::with_status(
+                                warp::reply(),
+                                warp::http::StatusCode::PARTIAL_CONTENT,
+                            ))
+                        } else {
+                            Ok(warp::reply::with_status(
+                                warp::reply(),
+                                warp::http::StatusCode::OK,
+                            ))
+                        }
+                    })
+                    .await
+                }
+            },
+        );
 
     // GET node/peers/{peer_id}
     let get_node_peers_by_id = eth_v1
@@ -2861,7 +2963,7 @@ pub fn serve<T: BeaconChainTypes>(
                             // It's reasonably likely that two different validators produce
                             // identical aggregates, especially if they're using the same beacon
                             // node.
-                            Err(AttnError::AttestationAlreadyKnown(_)) => continue,
+                            Err(AttnError::AttestationSupersetKnown(_)) => continue,
                             // If we've already seen this aggregator produce an aggregate, just
                             // skip this one.
                             //
@@ -3865,6 +3967,8 @@ pub fn serve<T: BeaconChainTypes>(
             warp::post().and(
                 post_beacon_blocks
                     .uor(post_beacon_blinded_blocks)
+                    .uor(post_beacon_blocks_v2)
+                    .uor(post_beacon_blinded_blocks_v2)
                     .uor(post_beacon_pool_attestations)
                     .uor(post_beacon_pool_attester_slashings)
                     .uor(post_beacon_pool_proposer_slashings)