From 2afa87879bd1b904e037a7ba67c001a5f1a162cc Mon Sep 17 00:00:00 2001 From: Jimmy Chen Date: Fri, 5 Dec 2025 21:59:42 +1100 Subject: [PATCH] Move beacon pool http api to its own separate module (#8543) Continuation of: * #8536 Moving `/beacon/pool` endpoints out of `http_api` to a separation module. This should improve code maintainability, incremental compilation time and rust analyzer performance. This is a tedious but straight forward change, so we're going with a pair & insta-merge approach to avoid painful & slow async review Co-Authored-By: Jimmy Chen --- beacon_node/http_api/src/beacon/mod.rs | 1 + beacon_node/http_api/src/beacon/pool.rs | 522 ++++++++++++++++++++++++ beacon_node/http_api/src/lib.rs | 459 ++------------------- beacon_node/http_api/src/utils.rs | 3 +- 4 files changed, 563 insertions(+), 422 deletions(-) create mode 100644 beacon_node/http_api/src/beacon/pool.rs diff --git a/beacon_node/http_api/src/beacon/mod.rs b/beacon_node/http_api/src/beacon/mod.rs index 20394784ae..df5e6eee5c 100644 --- a/beacon_node/http_api/src/beacon/mod.rs +++ b/beacon_node/http_api/src/beacon/mod.rs @@ -1 +1,2 @@ +pub mod pool; pub mod states; diff --git a/beacon_node/http_api/src/beacon/pool.rs b/beacon_node/http_api/src/beacon/pool.rs new file mode 100644 index 0000000000..059573c317 --- /dev/null +++ b/beacon_node/http_api/src/beacon/pool.rs @@ -0,0 +1,522 @@ +use crate::task_spawner::{Priority, TaskSpawner}; +use crate::utils::{NetworkTxFilter, OptionalConsensusVersionHeaderFilter, ResponseFilter}; +use crate::version::{ + ResponseIncludesVersion, V1, V2, add_consensus_version_header, beacon_response, + unsupported_version_rejection, +}; +use crate::{sync_committees, utils}; +use beacon_chain::observed_operations::ObservationOutcome; +use beacon_chain::{BeaconChain, BeaconChainTypes}; +use eth2::types::{AttestationPoolQuery, EndpointVersion, Failure, GenericResponse}; +use lighthouse_network::PubsubMessage; +use network::NetworkMessage; +use operation_pool::ReceivedPreCapella; +use slot_clock::SlotClock; +use std::collections::HashSet; +use std::sync::Arc; +use tokio::sync::mpsc::UnboundedSender; +use tracing::{debug, info, warn}; +use types::{ + Attestation, AttestationData, AttesterSlashing, ForkName, ProposerSlashing, + SignedBlsToExecutionChange, SignedVoluntaryExit, SingleAttestation, SyncCommitteeMessage, +}; +use warp::filters::BoxedFilter; +use warp::{Filter, Reply}; +use warp_utils::reject::convert_rejection; + +pub type BeaconPoolPathFilter = BoxedFilter<( + TaskSpawner<::EthSpec>, + Arc>, +)>; +pub type BeaconPoolPathV2Filter = BoxedFilter<( + TaskSpawner<::EthSpec>, + Arc>, +)>; +pub type BeaconPoolPathAnyFilter = BoxedFilter<( + EndpointVersion, + TaskSpawner<::EthSpec>, + Arc>, +)>; + +/// POST beacon/pool/bls_to_execution_changes +pub fn post_beacon_pool_bls_to_execution_changes( + network_tx_filter: &NetworkTxFilter, + beacon_pool_path: &BeaconPoolPathFilter, +) -> ResponseFilter { + beacon_pool_path + .clone() + .and(warp::path("bls_to_execution_changes")) + .and(warp::path::end()) + .and(warp_utils::json::json()) + .and(network_tx_filter.clone()) + .then( + |task_spawner: TaskSpawner, + chain: Arc>, + address_changes: Vec, + network_tx: UnboundedSender>| { + task_spawner.blocking_json_task(Priority::P0, 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) { + utils::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!( + %validator_index, + ?address, + published = + matches!(received_pre_capella, ReceivedPreCapella::No), + imported, + "Processed BLS to execution change" + ); + } + Ok(ObservationOutcome::AlreadyKnown) => { + debug!(%validator_index, "BLS to execution change already known"); + } + Err(e) => { + warn!( + validator_index, + reason = ?e, + source = "HTTP", + "Invalid BLS to execution change" + ); + failures.push(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, + )) + } + }) + }, + ) + .boxed() +} + +/// GET beacon/pool/bls_to_execution_changes +pub fn get_beacon_pool_bls_to_execution_changes( + beacon_pool_path: &BeaconPoolPathFilter, +) -> ResponseFilter { + beacon_pool_path + .clone() + .and(warp::path("bls_to_execution_changes")) + .and(warp::path::end()) + .then( + |task_spawner: TaskSpawner, chain: Arc>| { + task_spawner.blocking_json_task(Priority::P1, move || { + let address_changes = chain.op_pool.get_all_bls_to_execution_changes(); + Ok(GenericResponse::from(address_changes)) + }) + }, + ) + .boxed() +} + +/// POST beacon/pool/sync_committees +pub fn post_beacon_pool_sync_committees( + network_tx_filter: &NetworkTxFilter, + beacon_pool_path: &BeaconPoolPathFilter, +) -> ResponseFilter { + beacon_pool_path + .clone() + .and(warp::path("sync_committees")) + .and(warp::path::end()) + .and(warp_utils::json::json()) + .and(network_tx_filter.clone()) + .then( + |task_spawner: TaskSpawner, + chain: Arc>, + signatures: Vec, + network_tx: UnboundedSender>| { + task_spawner.blocking_json_task(Priority::P0, move || { + sync_committees::process_sync_committee_signatures( + signatures, network_tx, &chain, + )?; + Ok(GenericResponse::from(())) + }) + }, + ) + .boxed() +} + +/// GET beacon/pool/voluntary_exits +pub fn get_beacon_pool_voluntary_exits( + beacon_pool_path: &BeaconPoolPathFilter, +) -> ResponseFilter { + beacon_pool_path + .clone() + .and(warp::path("voluntary_exits")) + .and(warp::path::end()) + .then( + |task_spawner: TaskSpawner, chain: Arc>| { + task_spawner.blocking_json_task(Priority::P1, move || { + let attestations = chain.op_pool.get_all_voluntary_exits(); + Ok(GenericResponse::from(attestations)) + }) + }, + ) + .boxed() +} + +/// POST beacon/pool/voluntary_exits +pub fn post_beacon_pool_voluntary_exits( + network_tx_filter: &NetworkTxFilter, + beacon_pool_path: &BeaconPoolPathFilter, +) -> ResponseFilter { + beacon_pool_path + .clone() + .and(warp::path("voluntary_exits")) + .and(warp::path::end()) + .and(warp_utils::json::json()) + .and(network_tx_filter.clone()) + .then( + |task_spawner: TaskSpawner, + chain: Arc>, + exit: SignedVoluntaryExit, + network_tx: UnboundedSender>| { + task_spawner.blocking_json_task(Priority::P0, move || { + let outcome = chain + .verify_voluntary_exit_for_gossip(exit.clone()) + .map_err(|e| { + warp_utils::reject::object_invalid(format!( + "gossip verification failed: {:?}", + e + )) + })?; + + // Notify the validator monitor. + chain + .validator_monitor + .read() + .register_api_voluntary_exit(&exit.message); + + if let ObservationOutcome::New(exit) = outcome { + utils::publish_pubsub_message( + &network_tx, + PubsubMessage::VoluntaryExit(Box::new(exit.clone().into_inner())), + )?; + + chain.import_voluntary_exit(exit); + } + + Ok(()) + }) + }, + ) + .boxed() +} + +/// GET beacon/pool/proposer_slashings +pub fn get_beacon_pool_proposer_slashings( + beacon_pool_path: &BeaconPoolPathFilter, +) -> ResponseFilter { + beacon_pool_path + .clone() + .and(warp::path("proposer_slashings")) + .and(warp::path::end()) + .then( + |task_spawner: TaskSpawner, chain: Arc>| { + task_spawner.blocking_json_task(Priority::P1, move || { + let attestations = chain.op_pool.get_all_proposer_slashings(); + Ok(GenericResponse::from(attestations)) + }) + }, + ) + .boxed() +} + +/// POST beacon/pool/proposer_slashings +pub fn post_beacon_pool_proposer_slashings( + network_tx_filter: &NetworkTxFilter, + beacon_pool_path: &BeaconPoolPathFilter, +) -> ResponseFilter { + beacon_pool_path + .clone() + .and(warp::path("proposer_slashings")) + .and(warp::path::end()) + .and(warp_utils::json::json()) + .and(network_tx_filter.clone()) + .then( + |task_spawner: TaskSpawner, + chain: Arc>, + slashing: ProposerSlashing, + network_tx: UnboundedSender>| { + task_spawner.blocking_json_task(Priority::P0, move || { + let outcome = chain + .verify_proposer_slashing_for_gossip(slashing.clone()) + .map_err(|e| { + warp_utils::reject::object_invalid(format!( + "gossip verification failed: {:?}", + e + )) + })?; + + // Notify the validator monitor. + chain + .validator_monitor + .read() + .register_api_proposer_slashing(&slashing); + + if let ObservationOutcome::New(slashing) = outcome { + utils::publish_pubsub_message( + &network_tx, + PubsubMessage::ProposerSlashing(Box::new( + slashing.clone().into_inner(), + )), + )?; + + chain.import_proposer_slashing(slashing); + } + + Ok(()) + }) + }, + ) + .boxed() +} + +/// GET beacon/pool/attester_slashings +pub fn get_beacon_pool_attester_slashings( + beacon_pool_path_any: &BeaconPoolPathAnyFilter, +) -> ResponseFilter { + beacon_pool_path_any + .clone() + .and(warp::path("attester_slashings")) + .and(warp::path::end()) + .then( + |endpoint_version: EndpointVersion, + task_spawner: TaskSpawner, + chain: Arc>| { + task_spawner.blocking_response_task(Priority::P1, move || { + let slashings = chain.op_pool.get_all_attester_slashings(); + + // Use the current slot to find the fork version, and convert all messages to the + // current fork's format. This is to ensure consistent message types matching + // `Eth-Consensus-Version`. + let current_slot = + chain + .slot_clock + .now() + .ok_or(warp_utils::reject::custom_server_error( + "unable to read slot clock".to_string(), + ))?; + let fork_name = chain.spec.fork_name_at_slot::(current_slot); + let slashings = slashings + .into_iter() + .filter(|slashing| { + (fork_name.electra_enabled() + && matches!(slashing, AttesterSlashing::Electra(_))) + || (!fork_name.electra_enabled() + && matches!(slashing, AttesterSlashing::Base(_))) + }) + .collect::>(); + + let require_version = match endpoint_version { + V1 => ResponseIncludesVersion::No, + V2 => ResponseIncludesVersion::Yes(fork_name), + _ => return Err(unsupported_version_rejection(endpoint_version)), + }; + + let res = beacon_response(require_version, &slashings); + Ok(add_consensus_version_header( + warp::reply::json(&res).into_response(), + fork_name, + )) + }) + }, + ) + .boxed() +} + +// POST beacon/pool/attester_slashings +pub fn post_beacon_pool_attester_slashings( + network_tx_filter: &NetworkTxFilter, + beacon_pool_path_any: &BeaconPoolPathAnyFilter, +) -> ResponseFilter { + beacon_pool_path_any + .clone() + .and(warp::path("attester_slashings")) + .and(warp::path::end()) + .and(warp_utils::json::json()) + .and(network_tx_filter.clone()) + .then( + // V1 and V2 are identical except V2 has a consensus version header in the request. + // We only require this header for SSZ deserialization, which isn't supported for + // this endpoint presently. + |_endpoint_version: EndpointVersion, + task_spawner: TaskSpawner, + chain: Arc>, + slashing: AttesterSlashing, + network_tx: UnboundedSender>| { + task_spawner.blocking_json_task(Priority::P0, move || { + let outcome = chain + .verify_attester_slashing_for_gossip(slashing.clone()) + .map_err(|e| { + warp_utils::reject::object_invalid(format!( + "gossip verification failed: {:?}", + e + )) + })?; + + // Notify the validator monitor. + chain + .validator_monitor + .read() + .register_api_attester_slashing(slashing.to_ref()); + + if let ObservationOutcome::New(slashing) = outcome { + utils::publish_pubsub_message( + &network_tx, + PubsubMessage::AttesterSlashing(Box::new( + slashing.clone().into_inner(), + )), + )?; + + chain.import_attester_slashing(slashing); + } + + Ok(()) + }) + }, + ) + .boxed() +} + +/// GET beacon/pool/attestations?committee_index,slot +pub fn get_beacon_pool_attestations( + beacon_pool_path_any: &BeaconPoolPathAnyFilter, +) -> ResponseFilter { + beacon_pool_path_any + .clone() + .and(warp::path("attestations")) + .and(warp::path::end()) + .and(warp::query::()) + .then( + |endpoint_version: EndpointVersion, + task_spawner: TaskSpawner, + chain: Arc>, + query: AttestationPoolQuery| { + task_spawner.blocking_response_task(Priority::P1, move || { + let query_filter = |data: &AttestationData, committee_indices: HashSet| { + query.slot.is_none_or(|slot| slot == data.slot) + && query + .committee_index + .is_none_or(|index| committee_indices.contains(&index)) + }; + + let mut attestations = chain.op_pool.get_filtered_attestations(query_filter); + attestations.extend( + chain + .naive_aggregation_pool + .read() + .iter() + .filter(|&att| { + query_filter(att.data(), att.get_committee_indices_map()) + }) + .cloned(), + ); + // Use the current slot to find the fork version, and convert all messages to the + // current fork's format. This is to ensure consistent message types matching + // `Eth-Consensus-Version`. + let current_slot = + chain + .slot_clock + .now() + .ok_or(warp_utils::reject::custom_server_error( + "unable to read slot clock".to_string(), + ))?; + let fork_name = chain.spec.fork_name_at_slot::(current_slot); + let attestations = attestations + .into_iter() + .filter(|att| { + (fork_name.electra_enabled() && matches!(att, Attestation::Electra(_))) + || (!fork_name.electra_enabled() + && matches!(att, Attestation::Base(_))) + }) + .collect::>(); + + let require_version = match endpoint_version { + V1 => ResponseIncludesVersion::No, + V2 => ResponseIncludesVersion::Yes(fork_name), + _ => return Err(unsupported_version_rejection(endpoint_version)), + }; + + let res = beacon_response(require_version, &attestations); + Ok(add_consensus_version_header( + warp::reply::json(&res).into_response(), + fork_name, + )) + }) + }, + ) + .boxed() +} + +pub fn post_beacon_pool_attestations_v2( + network_tx_filter: &NetworkTxFilter, + optional_consensus_version_header_filter: OptionalConsensusVersionHeaderFilter, + beacon_pool_path_v2: &BeaconPoolPathV2Filter, +) -> ResponseFilter { + beacon_pool_path_v2 + .clone() + .and(warp::path("attestations")) + .and(warp::path::end()) + .and(warp_utils::json::json::>()) + .and(optional_consensus_version_header_filter) + .and(network_tx_filter.clone()) + .then( + |task_spawner: TaskSpawner, + chain: Arc>, + attestations: Vec, + _fork_name: Option, + network_tx: UnboundedSender>| async move { + let result = crate::publish_attestations::publish_attestations( + task_spawner, + chain, + attestations, + network_tx, + true, + ) + .await + .map(|()| warp::reply::json(&())); + convert_rejection(result).await + }, + ) + .boxed() +} diff --git a/beacon_node/http_api/src/lib.rs b/beacon_node/http_api/src/lib.rs index 3801933855..628b94a2a7 100644 --- a/beacon_node/http_api/src/lib.rs +++ b/beacon_node/http_api/src/lib.rs @@ -36,16 +36,14 @@ mod validator_inclusion; mod validators; mod version; +use crate::beacon::pool::*; use crate::light_client::{get_light_client_bootstrap, get_light_client_updates}; use crate::utils::{AnyVersionFilter, EthV1Filter}; use crate::validator::post_validator_liveness_epoch; use crate::validator::*; use crate::version::beacon_response; use beacon::states; -use beacon_chain::{ - BeaconChain, BeaconChainError, BeaconChainTypes, WhenSlotSkipped, - observed_operations::ObservationOutcome, -}; +use beacon_chain::{BeaconChain, BeaconChainError, BeaconChainTypes, WhenSlotSkipped}; use beacon_processor::BeaconProcessorSend; pub use block_id::BlockId; use builder_states::get_next_withdrawals; @@ -62,12 +60,10 @@ use health_metrics::observe::Observe; use lighthouse_network::Enr; use lighthouse_network::NetworkGlobals; use lighthouse_network::PeerId; -use lighthouse_network::PubsubMessage; use lighthouse_version::version_with_platform; use logging::{SSELoggingComponents, crit}; use network::{NetworkMessage, NetworkSenders}; use network_utils::enr_ext::EnrExt; -use operation_pool::ReceivedPreCapella; use parking_lot::RwLock; pub use publish_blocks::{ ProvenancedBlock, publish_blinded_block, publish_block, reconstruct_block, @@ -76,7 +72,6 @@ use serde::{Deserialize, Serialize}; use slot_clock::SlotClock; use ssz::Encode; pub use state_id::StateId; -use std::collections::HashSet; use std::future::Future; use std::net::{IpAddr, Ipv4Addr, SocketAddr}; use std::path::PathBuf; @@ -93,9 +88,8 @@ use tokio_stream::{ }; use tracing::{debug, info, warn}; use types::{ - Attestation, AttestationData, AttesterSlashing, BeaconStateError, Checkpoint, ConfigAndPreset, - Epoch, EthSpec, ForkName, Hash256, ProposerSlashing, SignedBlindedBeaconBlock, - SignedBlsToExecutionChange, SignedVoluntaryExit, SingleAttestation, Slot, SyncCommitteeMessage, + BeaconStateError, Checkpoint, ConfigAndPreset, Epoch, EthSpec, ForkName, Hash256, + SignedBlindedBeaconBlock, Slot, }; use version::{ ResponseIncludesVersion, V1, V2, add_consensus_version_header, add_ssz_content_type_header, @@ -106,7 +100,7 @@ use warp::Reply; use warp::hyper::Body; use warp::sse::Event; use warp::{Filter, Rejection, http::Response}; -use warp_utils::{query::multi_key_query, reject::convert_rejection, uor::UnifyingOrFilter}; +use warp_utils::{query::multi_key_query, uor::UnifyingOrFilter}; const API_PREFIX: &str = "eth"; @@ -804,10 +798,10 @@ pub fn serve( * beacon/blocks */ let consensus_version_header_filter = - warp::header::header::(CONSENSUS_VERSION_HEADER); + warp::header::header::(CONSENSUS_VERSION_HEADER).boxed(); let optional_consensus_version_header_filter = - warp::header::optional::(CONSENSUS_VERSION_HEADER); + warp::header::optional::(CONSENSUS_VERSION_HEADER).boxed(); // POST beacon/blocks let post_beacon_blocks = eth_v1 @@ -816,7 +810,7 @@ pub fn serve( .and(warp::path("blocks")) .and(warp::path::end()) .and(warp::body::json()) - .and(consensus_version_header_filter) + .and(consensus_version_header_filter.clone()) .and(task_spawner_filter.clone()) .and(chain_filter.clone()) .and(network_tx_filter.clone()) @@ -853,7 +847,7 @@ pub fn serve( .and(warp::path("blocks")) .and(warp::path::end()) .and(warp::body::bytes()) - .and(consensus_version_header_filter) + .and(consensus_version_header_filter.clone()) .and(task_spawner_filter.clone()) .and(chain_filter.clone()) .and(network_tx_filter.clone()) @@ -891,7 +885,7 @@ pub fn serve( .and(warp::query::()) .and(warp::path::end()) .and(warp::body::json()) - .and(consensus_version_header_filter) + .and(consensus_version_header_filter.clone()) .and(task_spawner_filter.clone()) .and(chain_filter.clone()) .and(network_tx_filter.clone()) @@ -931,7 +925,7 @@ pub fn serve( .and(warp::query::()) .and(warp::path::end()) .and(warp::body::bytes()) - .and(consensus_version_header_filter) + .and(consensus_version_header_filter.clone()) .and(task_spawner_filter.clone()) .and(chain_filter.clone()) .and(network_tx_filter.clone()) @@ -1408,444 +1402,67 @@ pub fn serve( .and(warp::path("beacon")) .and(warp::path("pool")) .and(task_spawner_filter.clone()) - .and(chain_filter.clone()); + .and(chain_filter.clone()) + .boxed(); let beacon_pool_path_v2 = eth_v2 .clone() .and(warp::path("beacon")) .and(warp::path("pool")) .and(task_spawner_filter.clone()) - .and(chain_filter.clone()); + .and(chain_filter.clone()) + .boxed(); let beacon_pool_path_any = any_version .clone() .and(warp::path("beacon")) .and(warp::path("pool")) .and(task_spawner_filter.clone()) - .and(chain_filter.clone()); + .and(chain_filter.clone()) + .boxed(); - let post_beacon_pool_attestations_v2 = beacon_pool_path_v2 - .clone() - .and(warp::path("attestations")) - .and(warp::path::end()) - .and(warp_utils::json::json::>()) - .and(optional_consensus_version_header_filter) - .and(network_tx_filter.clone()) - .then( - |task_spawner: TaskSpawner, - chain: Arc>, - attestations: Vec, - _fork_name: Option, - network_tx: UnboundedSender>| async move { - let result = crate::publish_attestations::publish_attestations( - task_spawner, - chain, - attestations, - network_tx, - true, - ) - .await - .map(|()| warp::reply::json(&())); - convert_rejection(result).await - }, - ); + let post_beacon_pool_attestations_v2 = post_beacon_pool_attestations_v2( + &network_tx_filter, + optional_consensus_version_header_filter, + &beacon_pool_path_v2, + ); // GET beacon/pool/attestations?committee_index,slot - let get_beacon_pool_attestations = beacon_pool_path_any - .clone() - .and(warp::path("attestations")) - .and(warp::path::end()) - .and(warp::query::()) - .then( - |endpoint_version: EndpointVersion, - task_spawner: TaskSpawner, - chain: Arc>, - query: api_types::AttestationPoolQuery| { - task_spawner.blocking_response_task(Priority::P1, move || { - let query_filter = |data: &AttestationData, committee_indices: HashSet| { - query.slot.is_none_or(|slot| slot == data.slot) - && query - .committee_index - .is_none_or(|index| committee_indices.contains(&index)) - }; - - let mut attestations = chain.op_pool.get_filtered_attestations(query_filter); - attestations.extend( - chain - .naive_aggregation_pool - .read() - .iter() - .filter(|&att| { - query_filter(att.data(), att.get_committee_indices_map()) - }) - .cloned(), - ); - // Use the current slot to find the fork version, and convert all messages to the - // current fork's format. This is to ensure consistent message types matching - // `Eth-Consensus-Version`. - let current_slot = - chain - .slot_clock - .now() - .ok_or(warp_utils::reject::custom_server_error( - "unable to read slot clock".to_string(), - ))?; - let fork_name = chain.spec.fork_name_at_slot::(current_slot); - let attestations = attestations - .into_iter() - .filter(|att| { - (fork_name.electra_enabled() && matches!(att, Attestation::Electra(_))) - || (!fork_name.electra_enabled() - && matches!(att, Attestation::Base(_))) - }) - .collect::>(); - - let require_version = match endpoint_version { - V1 => ResponseIncludesVersion::No, - V2 => ResponseIncludesVersion::Yes(fork_name), - _ => return Err(unsupported_version_rejection(endpoint_version)), - }; - - let res = beacon_response(require_version, &attestations); - Ok(add_consensus_version_header( - warp::reply::json(&res).into_response(), - fork_name, - )) - }) - }, - ); + let get_beacon_pool_attestations = get_beacon_pool_attestations(&beacon_pool_path_any); // POST beacon/pool/attester_slashings - let post_beacon_pool_attester_slashings = beacon_pool_path_any - .clone() - .and(warp::path("attester_slashings")) - .and(warp::path::end()) - .and(warp_utils::json::json()) - .and(network_tx_filter.clone()) - .then( - // V1 and V2 are identical except V2 has a consensus version header in the request. - // We only require this header for SSZ deserialization, which isn't supported for - // this endpoint presently. - |_endpoint_version: EndpointVersion, - task_spawner: TaskSpawner, - chain: Arc>, - slashing: AttesterSlashing, - network_tx: UnboundedSender>| { - task_spawner.blocking_json_task(Priority::P0, move || { - let outcome = chain - .verify_attester_slashing_for_gossip(slashing.clone()) - .map_err(|e| { - warp_utils::reject::object_invalid(format!( - "gossip verification failed: {:?}", - e - )) - })?; - - // Notify the validator monitor. - chain - .validator_monitor - .read() - .register_api_attester_slashing(slashing.to_ref()); - - if let ObservationOutcome::New(slashing) = outcome { - utils::publish_pubsub_message( - &network_tx, - PubsubMessage::AttesterSlashing(Box::new( - slashing.clone().into_inner(), - )), - )?; - - chain.import_attester_slashing(slashing); - } - - Ok(()) - }) - }, - ); + let post_beacon_pool_attester_slashings = + post_beacon_pool_attester_slashings(&network_tx_filter, &beacon_pool_path_any); // GET beacon/pool/attester_slashings let get_beacon_pool_attester_slashings = - beacon_pool_path_any - .clone() - .and(warp::path("attester_slashings")) - .and(warp::path::end()) - .then( - |endpoint_version: EndpointVersion, - task_spawner: TaskSpawner, - chain: Arc>| { - task_spawner.blocking_response_task(Priority::P1, move || { - let slashings = chain.op_pool.get_all_attester_slashings(); - - // Use the current slot to find the fork version, and convert all messages to the - // current fork's format. This is to ensure consistent message types matching - // `Eth-Consensus-Version`. - let current_slot = chain.slot_clock.now().ok_or( - warp_utils::reject::custom_server_error( - "unable to read slot clock".to_string(), - ), - )?; - let fork_name = chain.spec.fork_name_at_slot::(current_slot); - let slashings = slashings - .into_iter() - .filter(|slashing| { - (fork_name.electra_enabled() - && matches!(slashing, AttesterSlashing::Electra(_))) - || (!fork_name.electra_enabled() - && matches!(slashing, AttesterSlashing::Base(_))) - }) - .collect::>(); - - let require_version = match endpoint_version { - V1 => ResponseIncludesVersion::No, - V2 => ResponseIncludesVersion::Yes(fork_name), - _ => return Err(unsupported_version_rejection(endpoint_version)), - }; - - let res = beacon_response(require_version, &slashings); - Ok(add_consensus_version_header( - warp::reply::json(&res).into_response(), - fork_name, - )) - }) - }, - ); + get_beacon_pool_attester_slashings(&beacon_pool_path_any); // POST beacon/pool/proposer_slashings - let post_beacon_pool_proposer_slashings = beacon_pool_path - .clone() - .and(warp::path("proposer_slashings")) - .and(warp::path::end()) - .and(warp_utils::json::json()) - .and(network_tx_filter.clone()) - .then( - |task_spawner: TaskSpawner, - chain: Arc>, - slashing: ProposerSlashing, - network_tx: UnboundedSender>| { - task_spawner.blocking_json_task(Priority::P0, move || { - let outcome = chain - .verify_proposer_slashing_for_gossip(slashing.clone()) - .map_err(|e| { - warp_utils::reject::object_invalid(format!( - "gossip verification failed: {:?}", - e - )) - })?; - - // Notify the validator monitor. - chain - .validator_monitor - .read() - .register_api_proposer_slashing(&slashing); - - if let ObservationOutcome::New(slashing) = outcome { - utils::publish_pubsub_message( - &network_tx, - PubsubMessage::ProposerSlashing(Box::new( - slashing.clone().into_inner(), - )), - )?; - - chain.import_proposer_slashing(slashing); - } - - Ok(()) - }) - }, - ); + let post_beacon_pool_proposer_slashings = + post_beacon_pool_proposer_slashings(&network_tx_filter, &beacon_pool_path); // GET beacon/pool/proposer_slashings - let get_beacon_pool_proposer_slashings = beacon_pool_path - .clone() - .and(warp::path("proposer_slashings")) - .and(warp::path::end()) - .then( - |task_spawner: TaskSpawner, chain: Arc>| { - task_spawner.blocking_json_task(Priority::P1, move || { - let attestations = chain.op_pool.get_all_proposer_slashings(); - Ok(api_types::GenericResponse::from(attestations)) - }) - }, - ); + let get_beacon_pool_proposer_slashings = get_beacon_pool_proposer_slashings(&beacon_pool_path); // POST beacon/pool/voluntary_exits - let post_beacon_pool_voluntary_exits = beacon_pool_path - .clone() - .and(warp::path("voluntary_exits")) - .and(warp::path::end()) - .and(warp_utils::json::json()) - .and(network_tx_filter.clone()) - .then( - |task_spawner: TaskSpawner, - chain: Arc>, - exit: SignedVoluntaryExit, - network_tx: UnboundedSender>| { - task_spawner.blocking_json_task(Priority::P0, move || { - let outcome = chain - .verify_voluntary_exit_for_gossip(exit.clone()) - .map_err(|e| { - warp_utils::reject::object_invalid(format!( - "gossip verification failed: {:?}", - e - )) - })?; - - // Notify the validator monitor. - chain - .validator_monitor - .read() - .register_api_voluntary_exit(&exit.message); - - if let ObservationOutcome::New(exit) = outcome { - utils::publish_pubsub_message( - &network_tx, - PubsubMessage::VoluntaryExit(Box::new(exit.clone().into_inner())), - )?; - - chain.import_voluntary_exit(exit); - } - - Ok(()) - }) - }, - ); + let post_beacon_pool_voluntary_exits = + post_beacon_pool_voluntary_exits(&network_tx_filter, &beacon_pool_path); // GET beacon/pool/voluntary_exits - let get_beacon_pool_voluntary_exits = beacon_pool_path - .clone() - .and(warp::path("voluntary_exits")) - .and(warp::path::end()) - .then( - |task_spawner: TaskSpawner, chain: Arc>| { - task_spawner.blocking_json_task(Priority::P1, move || { - let attestations = chain.op_pool.get_all_voluntary_exits(); - Ok(api_types::GenericResponse::from(attestations)) - }) - }, - ); + let get_beacon_pool_voluntary_exits = get_beacon_pool_voluntary_exits(&beacon_pool_path); // POST beacon/pool/sync_committees - let post_beacon_pool_sync_committees = beacon_pool_path - .clone() - .and(warp::path("sync_committees")) - .and(warp::path::end()) - .and(warp_utils::json::json()) - .and(network_tx_filter.clone()) - .then( - |task_spawner: TaskSpawner, - chain: Arc>, - signatures: Vec, - network_tx: UnboundedSender>| { - task_spawner.blocking_json_task(Priority::P0, move || { - sync_committees::process_sync_committee_signatures( - signatures, network_tx, &chain, - )?; - Ok(api_types::GenericResponse::from(())) - }) - }, - ); + let post_beacon_pool_sync_committees = + post_beacon_pool_sync_committees(&network_tx_filter, &beacon_pool_path); // 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()) - .then( - |task_spawner: TaskSpawner, chain: Arc>| { - task_spawner.blocking_json_task(Priority::P1, move || { - let address_changes = chain.op_pool.get_all_bls_to_execution_changes(); - Ok(api_types::GenericResponse::from(address_changes)) - }) - }, - ); + let get_beacon_pool_bls_to_execution_changes = + get_beacon_pool_bls_to_execution_changes(&beacon_pool_path); // 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_utils::json::json()) - .and(network_tx_filter.clone()) - .then( - |task_spawner: TaskSpawner, - chain: Arc>, - address_changes: Vec, - network_tx: UnboundedSender>| { - task_spawner.blocking_json_task(Priority::P0, 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) { - utils::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!( - %validator_index, - ?address, - published = - matches!(received_pre_capella, ReceivedPreCapella::No), - imported, - "Processed BLS to execution change" - ); - } - Ok(ObservationOutcome::AlreadyKnown) => { - debug!(%validator_index, "BLS to execution change already known"); - } - Err(e) => { - warn!( - validator_index, - reason = ?e, - source = "HTTP", - "Invalid BLS to execution change" - ); - 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, - )) - } - }) - }, - ); + let post_beacon_pool_bls_to_execution_changes = + post_beacon_pool_bls_to_execution_changes(&network_tx_filter, &beacon_pool_path); let beacon_rewards_path = eth_v1 .clone() diff --git a/beacon_node/http_api/src/utils.rs b/beacon_node/http_api/src/utils.rs index a89780ba24..f2b859ebe5 100644 --- a/beacon_node/http_api/src/utils.rs +++ b/beacon_node/http_api/src/utils.rs @@ -7,7 +7,7 @@ use network::{NetworkMessage, ValidatorSubscriptionMessage}; use parking_lot::RwLock; use std::sync::Arc; use tokio::sync::mpsc::{Sender, UnboundedSender}; -use types::{ChainSpec, EthSpec}; +use types::{ChainSpec, EthSpec, ForkName}; use warp::Rejection; use warp::filters::BoxedFilter; @@ -20,6 +20,7 @@ pub type TaskSpawnerFilter = BoxedFilter<(TaskSpawner< pub type ValidatorSubscriptionTxFilter = BoxedFilter<(Sender,)>; pub type NetworkTxFilter = BoxedFilter<(UnboundedSender::EthSpec>>,)>; +pub type OptionalConsensusVersionHeaderFilter = BoxedFilter<(Option,)>; pub fn from_meta_data( meta_data: &RwLock>,