diff --git a/beacon_node/http_api/src/lib.rs b/beacon_node/http_api/src/lib.rs index 56ba86bb16..858d10c842 100644 --- a/beacon_node/http_api/src/lib.rs +++ b/beacon_node/http_api/src/lib.rs @@ -77,11 +77,7 @@ use warp::http::StatusCode; use warp::sse::Event; use warp::Reply; use warp::{http::Response, Filter}; -use warp_utils::{ - query::multi_key_query, - task::{blocking_json_task, blocking_response_task}, - uor::UnifyingOrFilter, -}; +use warp_utils::{query::multi_key_query, uor::UnifyingOrFilter}; const API_PREFIX: &str = "eth"; @@ -1270,7 +1266,7 @@ pub fn serve( chain: Arc>, network_tx: UnboundedSender>, log: Logger| { - task_spawner.async_task(Priority::P1, async move { + task_spawner.spawn_async_with_rejection(Priority::P1, async move { publish_blocks::publish_block( None, ProvenancedBlock::local(block), @@ -1291,35 +1287,39 @@ pub fn serve( .and(warp::query::()) .and(warp::path::end()) .and(warp::body::json()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) .and(network_tx_filter.clone()) .and(log_filter.clone()) .then( |validation_level: api_types::BroadcastValidationQuery, block: Arc>, + task_spawner: TaskSpawner, chain: Arc>, network_tx: UnboundedSender>, - 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(), - }, - } + log: Logger| { + task_spawner.spawn_async(Priority::P1, 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(), + }, + } + }) }, ); @@ -1333,23 +1333,27 @@ pub fn serve( .and(warp::path("blinded_blocks")) .and(warp::path::end()) .and(warp::body::json()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) .and(network_tx_filter.clone()) .and(log_filter.clone()) .and_then( |block: SignedBeaconBlock>, + task_spawner: TaskSpawner, chain: Arc>, network_tx: UnboundedSender>, - log: Logger| async move { - publish_blocks::publish_blinded_block( - block, - chain, - &network_tx, - log, - BroadcastValidation::default(), - ) - .await - .map(|()| warp::reply().into_response()) + log: Logger| { + task_spawner.spawn_async_with_rejection(Priority::P1, async move { + publish_blocks::publish_blinded_block( + block, + chain, + &network_tx, + log, + BroadcastValidation::default(), + ) + .await + .map(|()| warp::reply().into_response()) + }) }, ); @@ -1359,34 +1363,38 @@ pub fn serve( .and(warp::query::()) .and(warp::path::end()) .and(warp::body::json()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) .and(network_tx_filter.clone()) .and(log_filter.clone()) .then( |validation_level: api_types::BroadcastValidationQuery, block: SignedBeaconBlock>, + task_spawner: TaskSpawner, chain: Arc>, network_tx: UnboundedSender>, - 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(), - }, - } + log: Logger| { + task_spawner.spawn_async(Priority::P1, 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(), + }, + } + }) }, ); @@ -1400,12 +1408,14 @@ pub fn serve( .and(warp::path("beacon")) .and(warp::path("blocks")) .and(block_id_or_err) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()); let beacon_blocks_path_any = any_version .and(warp::path("beacon")) .and(warp::path("blocks")) .and(block_id_or_err) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()); // GET beacon/blocks/{block_id} @@ -1416,9 +1426,10 @@ pub fn serve( .and_then( |endpoint_version: EndpointVersion, block_id: BlockId, + task_spawner: TaskSpawner, chain: Arc>, accept_header: Option| { - async move { + task_spawner.spawn_async_with_rejection(Priority::P1, async move { let (block, execution_optimistic, finalized) = block_id.full_block(&chain).await?; let fork_name = block @@ -1446,7 +1457,7 @@ pub fn serve( .map(|res| warp::reply::json(&res).into_response()), } .map(|resp| add_consensus_version_header(resp, fork_name)) - } + }) }, ); @@ -1455,44 +1466,56 @@ pub fn serve( .clone() .and(warp::path("root")) .and(warp::path::end()) - .and_then(|block_id: BlockId, chain: Arc>| { - blocking_json_task(move || { - let (block, execution_optimistic, finalized) = block_id.blinded_block(&chain)?; - Ok(api_types::GenericResponse::from(api_types::RootData::from( - block.canonical_root(), - )) - .add_execution_optimistic_finalized(execution_optimistic, finalized)) - }) - }); + .and_then( + |block_id: BlockId, + task_spawner: TaskSpawner, + chain: Arc>| { + task_spawner.blocking_json_task(Priority::P1, move || { + let (block, execution_optimistic, finalized) = + block_id.blinded_block(&chain)?; + Ok(api_types::GenericResponse::from(api_types::RootData::from( + block.canonical_root(), + )) + .add_execution_optimistic_finalized(execution_optimistic, finalized)) + }) + }, + ); // GET beacon/blocks/{block_id}/attestations let get_beacon_block_attestations = beacon_blocks_path_v1 .clone() .and(warp::path("attestations")) .and(warp::path::end()) - .and_then(|block_id: BlockId, chain: Arc>| { - blocking_json_task(move || { - let (block, execution_optimistic, finalized) = block_id.blinded_block(&chain)?; - Ok( - api_types::GenericResponse::from(block.message().body().attestations().clone()) - .add_execution_optimistic_finalized(execution_optimistic, finalized), - ) - }) - }); + .and_then( + |block_id: BlockId, + task_spawner: TaskSpawner, + chain: Arc>| { + task_spawner.blocking_json_task(Priority::P1, move || { + let (block, execution_optimistic, finalized) = + block_id.blinded_block(&chain)?; + Ok(api_types::GenericResponse::from( + block.message().body().attestations().clone(), + ) + .add_execution_optimistic_finalized(execution_optimistic, finalized)) + }) + }, + ); // GET beacon/blinded_blocks/{block_id} let get_beacon_blinded_block = eth_v1 .and(warp::path("beacon")) .and(warp::path("blinded_blocks")) .and(block_id_or_err) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) .and(warp::path::end()) .and(warp::header::optional::("accept")) .and_then( |block_id: BlockId, + task_spawner: TaskSpawner, chain: Arc>, accept_header: Option| { - blocking_response_task(move || { + task_spawner.blocking_response_task(Priority::P1, move || { let (block, execution_optimistic, finalized) = block_id.blinded_block(&chain)?; let fork_name = block @@ -1534,6 +1557,7 @@ pub fn serve( let beacon_pool_path = eth_v1 .and(warp::path("beacon")) .and(warp::path("pool")) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()); // POST beacon/pool/attestations @@ -1545,11 +1569,12 @@ pub fn serve( .and(network_tx_filter.clone()) .and(log_filter.clone()) .and_then( - |chain: Arc>, + |task_spawner: TaskSpawner, + chain: Arc>, attestations: Vec>, network_tx: UnboundedSender>, log: Logger| { - blocking_json_task(move || { + task_spawner.blocking_json_task(Priority::P0, move || { let seen_timestamp = timestamp_now(); let mut failures = Vec::new(); let mut num_already_known = 0; @@ -1686,8 +1711,10 @@ pub fn serve( .and(warp::path::end()) .and(warp::query::()) .and_then( - |chain: Arc>, query: api_types::AttestationPoolQuery| { - blocking_json_task(move || { + |task_spawner: TaskSpawner, + chain: Arc>, + query: api_types::AttestationPoolQuery| { + task_spawner.blocking_json_task(Priority::P1, move || { let query_filter = |data: &AttestationData| { query.slot.map_or(true, |slot| slot == data.slot) && query @@ -1717,10 +1744,11 @@ pub fn serve( .and(warp::body::json()) .and(network_tx_filter.clone()) .and_then( - |chain: Arc>, + |task_spawner: TaskSpawner, + chain: Arc>, slashing: AttesterSlashing, network_tx: UnboundedSender>| { - blocking_json_task(move || { + task_spawner.blocking_json_task(Priority::P0, move || { let outcome = chain .verify_attester_slashing_for_gossip(slashing.clone()) .map_err(|e| { @@ -1757,12 +1785,14 @@ pub fn serve( .clone() .and(warp::path("attester_slashings")) .and(warp::path::end()) - .and_then(|chain: Arc>| { - blocking_json_task(move || { - let attestations = chain.op_pool.get_all_attester_slashings(); - Ok(api_types::GenericResponse::from(attestations)) - }) - }); + .and_then( + |task_spawner: TaskSpawner, chain: Arc>| { + task_spawner.blocking_json_task(Priority::P1, move || { + let attestations = chain.op_pool.get_all_attester_slashings(); + Ok(api_types::GenericResponse::from(attestations)) + }) + }, + ); // POST beacon/pool/proposer_slashings let post_beacon_pool_proposer_slashings = beacon_pool_path @@ -1772,10 +1802,11 @@ pub fn serve( .and(warp::body::json()) .and(network_tx_filter.clone()) .and_then( - |chain: Arc>, + |task_spawner: TaskSpawner, + chain: Arc>, slashing: ProposerSlashing, network_tx: UnboundedSender>| { - blocking_json_task(move || { + task_spawner.blocking_json_task(Priority::P0, move || { let outcome = chain .verify_proposer_slashing_for_gossip(slashing.clone()) .map_err(|e| { @@ -1812,12 +1843,14 @@ pub fn serve( .clone() .and(warp::path("proposer_slashings")) .and(warp::path::end()) - .and_then(|chain: Arc>| { - blocking_json_task(move || { - let attestations = chain.op_pool.get_all_proposer_slashings(); - Ok(api_types::GenericResponse::from(attestations)) - }) - }); + .and_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)) + }) + }, + ); // POST beacon/pool/voluntary_exits let post_beacon_pool_voluntary_exits = beacon_pool_path @@ -1827,10 +1860,11 @@ pub fn serve( .and(warp::body::json()) .and(network_tx_filter.clone()) .and_then( - |chain: Arc>, + |task_spawner: TaskSpawner, + chain: Arc>, exit: SignedVoluntaryExit, network_tx: UnboundedSender>| { - blocking_json_task(move || { + task_spawner.blocking_json_task(Priority::P0, move || { let outcome = chain .verify_voluntary_exit_for_gossip(exit.clone()) .map_err(|e| { @@ -1865,12 +1899,14 @@ pub fn serve( .clone() .and(warp::path("voluntary_exits")) .and(warp::path::end()) - .and_then(|chain: Arc>| { - blocking_json_task(move || { - let attestations = chain.op_pool.get_all_voluntary_exits(); - Ok(api_types::GenericResponse::from(attestations)) - }) - }); + .and_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)) + }) + }, + ); // POST beacon/pool/sync_committees let post_beacon_pool_sync_committees = beacon_pool_path @@ -1881,11 +1917,12 @@ pub fn serve( .and(network_tx_filter.clone()) .and(log_filter.clone()) .and_then( - |chain: Arc>, + |task_spawner: TaskSpawner, + chain: Arc>, signatures: Vec, network_tx: UnboundedSender>, log: Logger| { - blocking_json_task(move || { + task_spawner.blocking_json_task(Priority::P0, move || { sync_committees::process_sync_committee_signatures( signatures, network_tx, &chain, log, )?; @@ -1899,12 +1936,14 @@ pub fn serve( .clone() .and(warp::path("bls_to_execution_changes")) .and(warp::path::end()) - .and_then(|chain: Arc>| { - blocking_json_task(move || { - let address_changes = chain.op_pool.get_all_bls_to_execution_changes(); - Ok(api_types::GenericResponse::from(address_changes)) - }) - }); + .and_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)) + }) + }, + ); // POST beacon/pool/bls_to_execution_changes let post_beacon_pool_bls_to_execution_changes = beacon_pool_path @@ -1915,11 +1954,12 @@ pub fn serve( .and(network_tx_filter.clone()) .and(log_filter.clone()) .and_then( - |chain: Arc>, + |task_spawner: TaskSpawner, + chain: Arc>, address_changes: Vec, network_tx: UnboundedSender>, log: Logger| { - blocking_json_task(move || { + task_spawner.blocking_json_task(Priority::P0, move || { let mut failures = vec![]; for (index, address_change) in address_changes.into_iter().enumerate() { @@ -2003,10 +2043,13 @@ pub fn serve( .and(warp::path("deposit_snapshot")) .and(warp::path::end()) .and(warp::header::optional::("accept")) + .and(task_spawner_filter.clone()) .and(eth1_service_filter.clone()) .and_then( - |accept_header: Option, eth1_service: eth1::Service| { - blocking_response_task(move || match accept_header { + |accept_header: Option, + task_spawner: TaskSpawner, + eth1_service: eth1::Service| { + task_spawner.blocking_response_task(Priority::P1, move || match accept_header { Some(api_types::Accept::Json) | None => { let snapshot = eth1_service.get_deposit_snapshot(); Ok( @@ -2047,6 +2090,7 @@ pub fn serve( let beacon_rewards_path = eth_v1 .and(warp::path("beacon")) .and(warp::path("rewards")) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()); // GET beacon/rewards/blocks/{block_id} @@ -2055,17 +2099,21 @@ pub fn serve( .and(warp::path("blocks")) .and(block_id_or_err) .and(warp::path::end()) - .and_then(|chain: Arc>, block_id: BlockId| { - blocking_json_task(move || { - let (rewards, execution_optimistic, finalized) = - standard_block_rewards::compute_beacon_block_rewards(chain, block_id)?; - Ok(rewards) - .map(api_types::GenericResponse::from) - .map(|resp| { - resp.add_execution_optimistic_finalized(execution_optimistic, finalized) - }) - }) - }); + .and_then( + |task_spawner: TaskSpawner, + chain: Arc>, + block_id: BlockId| { + task_spawner.blocking_json_task(Priority::P1, move || { + let (rewards, execution_optimistic, finalized) = + standard_block_rewards::compute_beacon_block_rewards(chain, block_id)?; + Ok(rewards) + .map(api_types::GenericResponse::from) + .map(|resp| { + resp.add_execution_optimistic_finalized(execution_optimistic, finalized) + }) + }) + }, + ); /* * beacon/rewards @@ -2074,6 +2122,7 @@ pub fn serve( let beacon_rewards_path = eth_v1 .and(warp::path("beacon")) .and(warp::path("rewards")) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()); // POST beacon/rewards/attestations/{epoch} @@ -2085,11 +2134,12 @@ pub fn serve( .and(warp::body::json()) .and(log_filter.clone()) .and_then( - |chain: Arc>, + |task_spawner: TaskSpawner, + chain: Arc>, epoch: Epoch, validators: Vec, log: Logger| { - blocking_json_task(move || { + task_spawner.blocking_json_task(Priority::P1, move || { let attestation_rewards = chain .compute_attestation_rewards(epoch, validators, log) .map_err(|e| match e { @@ -2137,11 +2187,12 @@ pub fn serve( .and(warp::body::json()) .and(log_filter.clone()) .and_then( - |chain: Arc>, + |task_spawner: TaskSpawner, + chain: Arc>, block_id: BlockId, validators: Vec, log: Logger| { - blocking_json_task(move || { + task_spawner.blocking_json_task(Priority::P1, move || { let (rewards, execution_optimistic, finalized) = sync_committee_rewards::compute_sync_committee_rewards( chain, block_id, validators, log, @@ -2166,46 +2217,55 @@ pub fn serve( let get_config_fork_schedule = config_path .and(warp::path("fork_schedule")) .and(warp::path::end()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) - .and_then(|chain: Arc>| { - blocking_json_task(move || { - let forks = ForkName::list_all() - .into_iter() - .filter_map(|fork_name| chain.spec.fork_for_name(fork_name)) - .collect::>(); - Ok(api_types::GenericResponse::from(forks)) - }) - }); + .and_then( + |task_spawner: TaskSpawner, chain: Arc>| { + task_spawner.blocking_json_task(Priority::P1, move || { + let forks = ForkName::list_all() + .into_iter() + .filter_map(|fork_name| chain.spec.fork_for_name(fork_name)) + .collect::>(); + Ok(api_types::GenericResponse::from(forks)) + }) + }, + ); // GET config/spec let spec_fork_name = ctx.config.spec_fork_name; let get_config_spec = config_path .and(warp::path("spec")) .and(warp::path::end()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) - .and_then(move |chain: Arc>| { - blocking_json_task(move || { - let config_and_preset = - ConfigAndPreset::from_chain_spec::(&chain.spec, spec_fork_name); - Ok(api_types::GenericResponse::from(config_and_preset)) - }) - }); + .and_then( + move |task_spawner: TaskSpawner, chain: Arc>| { + task_spawner.blocking_json_task(Priority::P1, move || { + let config_and_preset = + ConfigAndPreset::from_chain_spec::(&chain.spec, spec_fork_name); + Ok(api_types::GenericResponse::from(config_and_preset)) + }) + }, + ); // GET config/deposit_contract let get_config_deposit_contract = config_path .and(warp::path("deposit_contract")) .and(warp::path::end()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) - .and_then(|chain: Arc>| { - blocking_json_task(move || { - Ok(api_types::GenericResponse::from( - api_types::DepositContractData { - address: chain.spec.deposit_contract_address, - chain_id: chain.spec.deposit_chain_id, - }, - )) - }) - }); + .and_then( + |task_spawner: TaskSpawner, chain: Arc>| { + task_spawner.blocking_json_task(Priority::P1, move || { + Ok(api_types::GenericResponse::from( + api_types::DepositContractData { + address: chain.spec.deposit_contract_address, + chain_id: chain.spec.deposit_chain_id, + }, + )) + }) + }, + ); /* * debug @@ -2223,13 +2283,15 @@ pub fn serve( })) .and(warp::path::end()) .and(warp::header::optional::("accept")) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) .and_then( |endpoint_version: EndpointVersion, state_id: StateId, accept_header: Option, + task_spawner: TaskSpawner, chain: Arc>| { - blocking_response_task(move || match accept_header { + task_spawner.blocking_response_task(Priority::P1, move || match accept_header { Some(api_types::Accept::Ssz) => { // We can ignore the optimistic status for the "fork" since it's a // specification constant that doesn't change across competing heads of the @@ -2281,10 +2343,13 @@ pub fn serve( .and(warp::path("beacon")) .and(warp::path("heads")) .and(warp::path::end()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) .and_then( - |endpoint_version: EndpointVersion, chain: Arc>| { - blocking_json_task(move || { + |endpoint_version: EndpointVersion, + task_spawner: TaskSpawner, + chain: Arc>| { + task_spawner.blocking_json_task(Priority::P1, move || { let heads = chain .heads() .into_iter() @@ -2317,48 +2382,51 @@ pub fn serve( .and(warp::path("debug")) .and(warp::path("fork_choice")) .and(warp::path::end()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) - .and_then(|chain: Arc>| { - blocking_json_task(move || { - let beacon_fork_choice = chain.canonical_head.fork_choice_read_lock(); + .and_then( + |task_spawner: TaskSpawner, chain: Arc>| { + task_spawner.blocking_json_task(Priority::P1, move || { + let beacon_fork_choice = chain.canonical_head.fork_choice_read_lock(); - let proto_array = beacon_fork_choice.proto_array().core_proto_array(); + let proto_array = beacon_fork_choice.proto_array().core_proto_array(); - let fork_choice_nodes = proto_array - .nodes - .iter() - .map(|node| { - let execution_status = if node.execution_status.is_execution_enabled() { - Some(node.execution_status.to_string()) - } else { - None - }; + let fork_choice_nodes = proto_array + .nodes + .iter() + .map(|node| { + let execution_status = if node.execution_status.is_execution_enabled() { + Some(node.execution_status.to_string()) + } else { + None + }; - ForkChoiceNode { - slot: node.slot, - block_root: node.root, - parent_root: node - .parent - .and_then(|index| proto_array.nodes.get(index)) - .map(|parent| parent.root), - justified_epoch: node.justified_checkpoint.epoch, - finalized_epoch: node.finalized_checkpoint.epoch, - weight: node.weight, - validity: execution_status, - execution_block_hash: node - .execution_status - .block_hash() - .map(|block_hash| block_hash.into_root()), - } + ForkChoiceNode { + slot: node.slot, + block_root: node.root, + parent_root: node + .parent + .and_then(|index| proto_array.nodes.get(index)) + .map(|parent| parent.root), + justified_epoch: node.justified_checkpoint.epoch, + finalized_epoch: node.finalized_checkpoint.epoch, + weight: node.weight, + validity: execution_status, + execution_block_hash: node + .execution_status + .block_hash() + .map(|block_hash| block_hash.into_root()), + } + }) + .collect::>(); + Ok(ForkChoice { + justified_checkpoint: proto_array.justified_checkpoint, + finalized_checkpoint: proto_array.finalized_checkpoint, + fork_choice_nodes, }) - .collect::>(); - Ok(ForkChoice { - justified_checkpoint: proto_array.justified_checkpoint, - finalized_checkpoint: proto_array.finalized_checkpoint, - fork_choice_nodes, }) - }) - }); + }, + ); /* * node @@ -2369,46 +2437,51 @@ pub fn serve( .and(warp::path("node")) .and(warp::path("identity")) .and(warp::path::end()) + .and(task_spawner_filter.clone()) .and(network_globals.clone()) - .and_then(|network_globals: Arc>| { - blocking_json_task(move || { - let enr = network_globals.local_enr(); - let p2p_addresses = enr.multiaddr_p2p_tcp(); - let discovery_addresses = enr.multiaddr_p2p_udp(); - let meta_data = network_globals.local_metadata.read(); - Ok(api_types::GenericResponse::from(api_types::IdentityData { - peer_id: network_globals.local_peer_id().to_base58(), - enr, - p2p_addresses, - discovery_addresses, - metadata: api_types::MetaData { - seq_number: *meta_data.seq_number(), - attnets: format!( - "0x{}", - hex::encode(meta_data.attnets().clone().into_bytes()), - ), - syncnets: format!( - "0x{}", - hex::encode( - meta_data - .syncnets() - .map(|x| x.clone()) - .unwrap_or_default() - .into_bytes() - ) - ), - }, - })) - }) - }); + .and_then( + |task_spawner: TaskSpawner, + network_globals: Arc>| { + task_spawner.blocking_json_task(Priority::P1, move || { + let enr = network_globals.local_enr(); + let p2p_addresses = enr.multiaddr_p2p_tcp(); + let discovery_addresses = enr.multiaddr_p2p_udp(); + let meta_data = network_globals.local_metadata.read(); + Ok(api_types::GenericResponse::from(api_types::IdentityData { + peer_id: network_globals.local_peer_id().to_base58(), + enr, + p2p_addresses, + discovery_addresses, + metadata: api_types::MetaData { + seq_number: *meta_data.seq_number(), + attnets: format!( + "0x{}", + hex::encode(meta_data.attnets().clone().into_bytes()), + ), + syncnets: format!( + "0x{}", + hex::encode( + meta_data + .syncnets() + .map(|x| x.clone()) + .unwrap_or_default() + .into_bytes() + ) + ), + }, + })) + }) + }, + ); // GET node/version let get_node_version = eth_v1 .and(warp::path("node")) .and(warp::path("version")) .and(warp::path::end()) - .and_then(|| { - blocking_json_task(move || { + .and(task_spawner_filter.clone()) + .and_then(|task_spawner: TaskSpawner| { + task_spawner.blocking_json_task(Priority::P1, move || { Ok(api_types::GenericResponse::from(api_types::VersionData { version: version_with_platform(), })) @@ -2420,10 +2493,13 @@ pub fn serve( .and(warp::path("node")) .and(warp::path("syncing")) .and(warp::path::end()) + .and(task_spawner_filter.clone()) .and(network_globals.clone()) .and(chain_filter.clone()) .and_then( - |network_globals: Arc>, chain: Arc>| { + |task_spawner: TaskSpawner, + network_globals: Arc>, + chain: Arc>| { async move { let el_offline = if let Some(el) = &chain.execution_layer { el.is_offline_or_erroring().await @@ -2431,32 +2507,34 @@ pub fn serve( true }; - blocking_json_task(move || { - let head_slot = chain.canonical_head.cached_head().head_slot(); - let current_slot = chain.slot_clock.now_or_genesis().ok_or_else(|| { - warp_utils::reject::custom_server_error( - "Unable to read slot clock".into(), - ) - })?; + task_spawner + .blocking_json_task(Priority::P1, move || { + let head_slot = chain.canonical_head.cached_head().head_slot(); + let current_slot = + chain.slot_clock.now_or_genesis().ok_or_else(|| { + warp_utils::reject::custom_server_error( + "Unable to read slot clock".into(), + ) + })?; - // Taking advantage of saturating subtraction on slot. - let sync_distance = current_slot - head_slot; + // Taking advantage of saturating subtraction on slot. + let sync_distance = current_slot - head_slot; - let is_optimistic = chain - .is_optimistic_or_invalid_head() - .map_err(warp_utils::reject::beacon_chain_error)?; + let is_optimistic = chain + .is_optimistic_or_invalid_head() + .map_err(warp_utils::reject::beacon_chain_error)?; - let syncing_data = api_types::SyncingData { - is_syncing: network_globals.sync_state.read().is_syncing(), - is_optimistic: Some(is_optimistic), - el_offline: Some(el_offline), - head_slot, - sync_distance, - }; + let syncing_data = api_types::SyncingData { + is_syncing: network_globals.sync_state.read().is_syncing(), + is_optimistic: Some(is_optimistic), + el_offline: Some(el_offline), + head_slot, + sync_distance, + }; - Ok(api_types::GenericResponse::from(syncing_data)) - }) - .await + Ok(api_types::GenericResponse::from(syncing_data)) + }) + .await } }, ); @@ -2466,10 +2544,13 @@ pub fn serve( .and(warp::path("node")) .and(warp::path("health")) .and(warp::path::end()) + .and(task_spawner_filter.clone()) .and(network_globals.clone()) .and(chain_filter.clone()) .and_then( - |network_globals: Arc>, chain: Arc>| { + |task_spawner: TaskSpawner, + network_globals: Arc>, + chain: Arc>| { async move { let el_offline = if let Some(el) = &chain.execution_layer { el.is_offline_or_erroring().await @@ -2477,28 +2558,31 @@ pub fn serve( true }; - blocking_response_task(move || { - let is_optimistic = chain - .is_optimistic_or_invalid_head() - .map_err(warp_utils::reject::beacon_chain_error)?; + task_spawner + .blocking_response_task(Priority::P1, 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(); + 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 + 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 } }, ); @@ -2509,10 +2593,13 @@ pub fn serve( .and(warp::path("peers")) .and(warp::path::param::()) .and(warp::path::end()) + .and(task_spawner_filter.clone()) .and(network_globals.clone()) .and_then( - |requested_peer_id: String, network_globals: Arc>| { - blocking_json_task(move || { + |requested_peer_id: String, + task_spawner: TaskSpawner, + network_globals: Arc>| { + task_spawner.blocking_json_task(Priority::P1, move || { let peer_id = PeerId::from_bytes( &bs58::decode(requested_peer_id.as_str()) .into_vec() @@ -2566,11 +2653,13 @@ pub fn serve( .and(warp::path("peers")) .and(warp::path::end()) .and(multi_key_query::()) + .and(task_spawner_filter.clone()) .and(network_globals.clone()) .and_then( |query_res: Result, + task_spawner: TaskSpawner, network_globals: Arc>| { - blocking_json_task(move || { + task_spawner.blocking_json_task(Priority::P1, move || { let query = query_res?; let mut peers: Vec = Vec::new(); network_globals @@ -2634,38 +2723,42 @@ pub fn serve( .and(warp::path("node")) .and(warp::path("peer_count")) .and(warp::path::end()) + .and(task_spawner_filter.clone()) .and(network_globals.clone()) - .and_then(|network_globals: Arc>| { - blocking_json_task(move || { - let mut connected: u64 = 0; - let mut connecting: u64 = 0; - let mut disconnected: u64 = 0; - let mut disconnecting: u64 = 0; + .and_then( + |task_spawner: TaskSpawner, + network_globals: Arc>| { + task_spawner.blocking_json_task(Priority::P1, move || { + let mut connected: u64 = 0; + let mut connecting: u64 = 0; + let mut disconnected: u64 = 0; + let mut disconnecting: u64 = 0; - network_globals - .peers - .read() - .peers() - .for_each(|(_, peer_info)| { - let state = api_types::PeerState::from_peer_connection_status( - peer_info.connection_status(), - ); - match state { - api_types::PeerState::Connected => connected += 1, - api_types::PeerState::Connecting => connecting += 1, - api_types::PeerState::Disconnected => disconnected += 1, - api_types::PeerState::Disconnecting => disconnecting += 1, - } - }); + network_globals + .peers + .read() + .peers() + .for_each(|(_, peer_info)| { + let state = api_types::PeerState::from_peer_connection_status( + peer_info.connection_status(), + ); + match state { + api_types::PeerState::Connected => connected += 1, + api_types::PeerState::Connecting => connecting += 1, + api_types::PeerState::Disconnected => disconnected += 1, + api_types::PeerState::Disconnecting => disconnecting += 1, + } + }); - Ok(api_types::GenericResponse::from(api_types::PeerCount { - connected, - connecting, - disconnected, - disconnecting, - })) - }) - }); + Ok(api_types::GenericResponse::from(api_types::PeerCount { + connected, + connecting, + disconnected, + disconnecting, + })) + }) + }, + ); /* * validator */ @@ -2682,11 +2775,19 @@ pub fn serve( })) .and(warp::path::end()) .and(not_while_syncing_filter.clone()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) .and(log_filter.clone()) - .and_then(|epoch: Epoch, chain: Arc>, log: Logger| { - blocking_json_task(move || proposer_duties::proposer_duties(epoch, &chain, &log)) - }); + .and_then( + |epoch: Epoch, + task_spawner: TaskSpawner, + chain: Arc>, + log: Logger| { + task_spawner.blocking_json_task(Priority::P0, move || { + proposer_duties::proposer_duties(epoch, &chain, &log) + }) + }, + ); // GET validator/blocks/{slot} let get_validator_blocks = any_version @@ -2700,56 +2801,60 @@ pub fn serve( .and(warp::path::end()) .and(not_while_syncing_filter.clone()) .and(warp::query::()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) .and(log_filter.clone()) .and_then( |endpoint_version: EndpointVersion, slot: Slot, query: api_types::ValidatorBlocksQuery, + task_spawner: TaskSpawner, chain: Arc>, - log: Logger| async move { - debug!( - log, - "Block production request from HTTP API"; - "slot" => slot - ); + log: Logger| { + task_spawner.spawn_async_with_rejection(Priority::P0, async move { + debug!( + log, + "Block production request from HTTP API"; + "slot" => slot + ); - let randao_reveal = query.randao_reveal.decompress().map_err(|e| { - warp_utils::reject::custom_bad_request(format!( - "randao reveal is not a valid BLS signature: {:?}", - e - )) - })?; + let randao_reveal = query.randao_reveal.decompress().map_err(|e| { + warp_utils::reject::custom_bad_request(format!( + "randao reveal is not a valid BLS signature: {:?}", + e + )) + })?; - let randao_verification = - if query.skip_randao_verification == SkipRandaoVerification::Yes { - if !randao_reveal.is_infinity() { - return Err(warp_utils::reject::custom_bad_request( + let randao_verification = + if query.skip_randao_verification == SkipRandaoVerification::Yes { + if !randao_reveal.is_infinity() { + return Err(warp_utils::reject::custom_bad_request( "randao_reveal must be point-at-infinity if verification is skipped" .into(), )); - } - ProduceBlockVerification::NoVerification - } else { - ProduceBlockVerification::VerifyRandao - }; + } + ProduceBlockVerification::NoVerification + } else { + ProduceBlockVerification::VerifyRandao + }; - let (block, _) = chain - .produce_block_with_verification::>( - randao_reveal, - slot, - query.graffiti.map(Into::into), - randao_verification, - ) - .await - .map_err(warp_utils::reject::block_production_error)?; - let fork_name = block - .to_ref() - .fork_name(&chain.spec) - .map_err(inconsistent_fork_rejection)?; + let (block, _) = chain + .produce_block_with_verification::>( + randao_reveal, + slot, + query.graffiti.map(Into::into), + randao_verification, + ) + .await + .map_err(warp_utils::reject::block_production_error)?; + let fork_name = block + .to_ref() + .fork_name(&chain.spec) + .map_err(inconsistent_fork_rejection)?; - fork_versioned_response(endpoint_version, fork_name, block) - .map(|response| warp::reply::json(&response).into_response()) + fork_versioned_response(endpoint_version, fork_name, block) + .map(|response| warp::reply::json(&response).into_response()) + }) }, ); @@ -2765,48 +2870,52 @@ pub fn serve( .and(warp::path::end()) .and(not_while_syncing_filter.clone()) .and(warp::query::()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) .and_then( |slot: Slot, query: api_types::ValidatorBlocksQuery, - chain: Arc>| async move { - let randao_reveal = query.randao_reveal.decompress().map_err(|e| { - warp_utils::reject::custom_bad_request(format!( - "randao reveal is not a valid BLS signature: {:?}", - e - )) - })?; + task_spawner: TaskSpawner, + chain: Arc>| { + task_spawner.spawn_async_with_rejection(Priority::P1, async move { + let randao_reveal = query.randao_reveal.decompress().map_err(|e| { + warp_utils::reject::custom_bad_request(format!( + "randao reveal is not a valid BLS signature: {:?}", + e + )) + })?; - let randao_verification = - if query.skip_randao_verification == SkipRandaoVerification::Yes { - if !randao_reveal.is_infinity() { - return Err(warp_utils::reject::custom_bad_request( + let randao_verification = + if query.skip_randao_verification == SkipRandaoVerification::Yes { + if !randao_reveal.is_infinity() { + return Err(warp_utils::reject::custom_bad_request( "randao_reveal must be point-at-infinity if verification is skipped" .into() )); - } - ProduceBlockVerification::NoVerification - } else { - ProduceBlockVerification::VerifyRandao - }; + } + ProduceBlockVerification::NoVerification + } else { + ProduceBlockVerification::VerifyRandao + }; - let (block, _) = chain - .produce_block_with_verification::>( - randao_reveal, - slot, - query.graffiti.map(Into::into), - randao_verification, - ) - .await - .map_err(warp_utils::reject::block_production_error)?; - let fork_name = block - .to_ref() - .fork_name(&chain.spec) - .map_err(inconsistent_fork_rejection)?; + let (block, _) = chain + .produce_block_with_verification::>( + randao_reveal, + slot, + query.graffiti.map(Into::into), + randao_verification, + ) + .await + .map_err(warp_utils::reject::block_production_error)?; + let fork_name = block + .to_ref() + .fork_name(&chain.spec) + .map_err(inconsistent_fork_rejection)?; - // Pose as a V2 endpoint so we return the fork `version`. - fork_versioned_response(V2, fork_name, block) - .map(|response| warp::reply::json(&response).into_response()) + // Pose as a V2 endpoint so we return the fork `version`. + fork_versioned_response(V2, fork_name, block) + .map(|response| warp::reply::json(&response).into_response()) + }) }, ); @@ -2817,10 +2926,13 @@ pub fn serve( .and(warp::path::end()) .and(warp::query::()) .and(not_while_syncing_filter.clone()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) .and_then( - |query: api_types::ValidatorAttestationDataQuery, chain: Arc>| { - blocking_json_task(move || { + |query: api_types::ValidatorAttestationDataQuery, + task_spawner: TaskSpawner, + chain: Arc>| { + task_spawner.blocking_json_task(Priority::P0, move || { let current_slot = chain .slot() .map_err(warp_utils::reject::beacon_chain_error)?; @@ -2849,10 +2961,13 @@ pub fn serve( .and(warp::path::end()) .and(warp::query::()) .and(not_while_syncing_filter.clone()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) .and_then( - |query: api_types::ValidatorAggregateAttestationQuery, chain: Arc>| { - blocking_json_task(move || { + |query: api_types::ValidatorAggregateAttestationQuery, + task_spawner: TaskSpawner, + chain: Arc>| { + task_spawner.blocking_json_task(Priority::P0, move || { chain .get_aggregated_attestation_by_slot_and_root( query.slot, @@ -2887,10 +3002,14 @@ pub fn serve( .and(warp::path::end()) .and(not_while_syncing_filter.clone()) .and(warp::body::json()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) .and_then( - |epoch: Epoch, indices: api_types::ValidatorIndexData, chain: Arc>| { - blocking_json_task(move || { + |epoch: Epoch, + indices: api_types::ValidatorIndexData, + task_spawner: TaskSpawner, + chain: Arc>| { + task_spawner.blocking_json_task(Priority::P0, move || { attester_duties::attester_duties(epoch, &indices.0, &chain) }) }, @@ -2909,10 +3028,14 @@ pub fn serve( .and(warp::path::end()) .and(not_while_syncing_filter.clone()) .and(warp::body::json()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) .and_then( - |epoch: Epoch, indices: api_types::ValidatorIndexData, chain: Arc>| { - blocking_json_task(move || { + |epoch: Epoch, + indices: api_types::ValidatorIndexData, + task_spawner: TaskSpawner, + chain: Arc>| { + task_spawner.blocking_json_task(Priority::P0, move || { sync_committees::sync_committee_duties(epoch, &indices.0, &chain) }) }, @@ -2925,10 +3048,13 @@ pub fn serve( .and(warp::path::end()) .and(warp::query::()) .and(not_while_syncing_filter.clone()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) .and_then( - |sync_committee_data: SyncContributionData, chain: Arc>| { - blocking_json_task(move || { + |sync_committee_data: SyncContributionData, + task_spawner: TaskSpawner, + chain: Arc>| { + task_spawner.blocking_json_task(Priority::P0, move || { chain .get_aggregated_sync_committee_contribution(&sync_committee_data) .map_err(|e| { @@ -2953,15 +3079,17 @@ pub fn serve( .and(warp::path("aggregate_and_proofs")) .and(warp::path::end()) .and(not_while_syncing_filter.clone()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) .and(warp::body::json()) .and(network_tx_filter.clone()) .and(log_filter.clone()) .and_then( - |chain: Arc>, + |task_spawner: TaskSpawner, + chain: Arc>, aggregates: Vec>, network_tx: UnboundedSender>, log: Logger| { - blocking_json_task(move || { + task_spawner.blocking_json_task(Priority::P0, move || { let seen_timestamp = timestamp_now(); let mut verified_aggregates = Vec::with_capacity(aggregates.len()); let mut messages = Vec::with_capacity(aggregates.len()); @@ -3064,16 +3192,18 @@ pub fn serve( .and(warp::path("contribution_and_proofs")) .and(warp::path::end()) .and(not_while_syncing_filter.clone()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) .and(warp::body::json()) .and(network_tx_filter) .and(log_filter.clone()) .and_then( - |chain: Arc>, + |task_spawner: TaskSpawner, + chain: Arc>, contributions: Vec>, network_tx: UnboundedSender>, log: Logger| { - blocking_json_task(move || { + task_spawner.blocking_json_task(Priority::P0, move || { sync_committees::process_signed_contribution_and_proofs( contributions, network_tx, @@ -3092,14 +3222,16 @@ pub fn serve( .and(warp::path::end()) .and(warp::body::json()) .and(validator_subscription_tx_filter.clone()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) .and(log_filter.clone()) .and_then( |subscriptions: Vec, validator_subscription_tx: Sender, + task_spawner: TaskSpawner, chain: Arc>, log: Logger| { - blocking_json_task(move || { + task_spawner.blocking_json_task(Priority::P0, move || { for subscription in &subscriptions { chain .validator_monitor @@ -3141,45 +3273,49 @@ pub fn serve( .and(warp::path("prepare_beacon_proposer")) .and(warp::path::end()) .and(not_while_syncing_filter.clone()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) .and(log_filter.clone()) .and(warp::body::json()) .and_then( - |chain: Arc>, + |task_spawner: TaskSpawner, + chain: Arc>, log: Logger, - preparation_data: Vec| async move { - let execution_layer = chain - .execution_layer - .as_ref() - .ok_or(BeaconChainError::ExecutionLayerMissing) - .map_err(warp_utils::reject::beacon_chain_error)?; + preparation_data: Vec| { + task_spawner.spawn_async_with_rejection(Priority::P0, async move { + let execution_layer = chain + .execution_layer + .as_ref() + .ok_or(BeaconChainError::ExecutionLayerMissing) + .map_err(warp_utils::reject::beacon_chain_error)?; - let current_slot = chain - .slot() - .map_err(warp_utils::reject::beacon_chain_error)?; - let current_epoch = current_slot.epoch(T::EthSpec::slots_per_epoch()); + let current_slot = chain + .slot() + .map_err(warp_utils::reject::beacon_chain_error)?; + let current_epoch = current_slot.epoch(T::EthSpec::slots_per_epoch()); - debug!( - log, - "Received proposer preparation data"; - "count" => preparation_data.len(), - ); + debug!( + log, + "Received proposer preparation data"; + "count" => preparation_data.len(), + ); - execution_layer - .update_proposer_preparation(current_epoch, &preparation_data) - .await; + execution_layer + .update_proposer_preparation(current_epoch, &preparation_data) + .await; - chain - .prepare_beacon_proposer(current_slot) - .await - .map_err(|e| { - warp_utils::reject::custom_bad_request(format!( - "error updating proposer preparations: {:?}", - e - )) - })?; + chain + .prepare_beacon_proposer(current_slot) + .await + .map_err(|e| { + warp_utils::reject::custom_bad_request(format!( + "error updating proposer preparations: {:?}", + e + )) + })?; - Ok::<_, warp::reject::Rejection>(warp::reply::json(&()).into_response()) + Ok::<_, warp::reject::Rejection>(warp::reply::json(&()).into_response()) + }) }, ); @@ -3188,127 +3324,133 @@ pub fn serve( .and(warp::path("validator")) .and(warp::path("register_validator")) .and(warp::path::end()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) .and(log_filter.clone()) .and(warp::body::json()) .and_then( - |chain: Arc>, + |task_spawner: TaskSpawner, + chain: Arc>, log: Logger, - register_val_data: Vec| async move { - let execution_layer = chain - .execution_layer - .as_ref() - .ok_or(BeaconChainError::ExecutionLayerMissing) - .map_err(warp_utils::reject::beacon_chain_error)?; - let current_slot = chain - .slot_clock - .now_or_genesis() - .ok_or(BeaconChainError::UnableToReadSlot) - .map_err(warp_utils::reject::beacon_chain_error)?; - let current_epoch = current_slot.epoch(T::EthSpec::slots_per_epoch()); + register_val_data: Vec| { + task_spawner.spawn_async_with_rejection(Priority::P0, async move { + let execution_layer = chain + .execution_layer + .as_ref() + .ok_or(BeaconChainError::ExecutionLayerMissing) + .map_err(warp_utils::reject::beacon_chain_error)?; + let current_slot = chain + .slot_clock + .now_or_genesis() + .ok_or(BeaconChainError::UnableToReadSlot) + .map_err(warp_utils::reject::beacon_chain_error)?; + let current_epoch = current_slot.epoch(T::EthSpec::slots_per_epoch()); - debug!( - log, - "Received register validator request"; - "count" => register_val_data.len(), - ); + debug!( + log, + "Received register validator request"; + "count" => register_val_data.len(), + ); - let head_snapshot = chain.head_snapshot(); - let spec = &chain.spec; + let head_snapshot = chain.head_snapshot(); + let spec = &chain.spec; - let (preparation_data, filtered_registration_data): ( - Vec, - Vec, - ) = register_val_data - .into_iter() - .filter_map(|register_data| { - chain - .validator_index(®ister_data.message.pubkey) - .ok() - .flatten() - .and_then(|validator_index| { - let validator = head_snapshot - .beacon_state - .get_validator(validator_index) - .ok()?; - let validator_status = ValidatorStatus::from_validator( - validator, - current_epoch, - spec.far_future_epoch, - ) - .superstatus(); - let is_active_or_pending = - matches!(validator_status, ValidatorStatus::Pending) - || matches!(validator_status, ValidatorStatus::Active); - - // Filter out validators who are not 'active' or 'pending'. - is_active_or_pending.then_some({ - ( - ProposerPreparationData { - validator_index: validator_index as u64, - fee_recipient: register_data.message.fee_recipient, - }, - register_data, + let (preparation_data, filtered_registration_data): ( + Vec, + Vec, + ) = register_val_data + .into_iter() + .filter_map(|register_data| { + chain + .validator_index(®ister_data.message.pubkey) + .ok() + .flatten() + .and_then(|validator_index| { + let validator = head_snapshot + .beacon_state + .get_validator(validator_index) + .ok()?; + let validator_status = ValidatorStatus::from_validator( + validator, + current_epoch, + spec.far_future_epoch, ) + .superstatus(); + let is_active_or_pending = + matches!(validator_status, ValidatorStatus::Pending) + || matches!(validator_status, ValidatorStatus::Active); + + // Filter out validators who are not 'active' or 'pending'. + is_active_or_pending.then_some({ + ( + ProposerPreparationData { + validator_index: validator_index as u64, + fee_recipient: register_data.message.fee_recipient, + }, + register_data, + ) + }) }) - }) - }) - .unzip(); + }) + .unzip(); - // Update the prepare beacon proposer cache based on this request. - execution_layer - .update_proposer_preparation(current_epoch, &preparation_data) - .await; + // Update the prepare beacon proposer cache based on this request. + execution_layer + .update_proposer_preparation(current_epoch, &preparation_data) + .await; - // Call prepare beacon proposer blocking with the latest update in order to make - // sure we have a local payload to fall back to in the event of the blinded block - // flow failing. - chain - .prepare_beacon_proposer(current_slot) - .await - .map_err(|e| { - warp_utils::reject::custom_bad_request(format!( - "error updating proposer preparations: {:?}", - e - )) - })?; + // Call prepare beacon proposer blocking with the latest update in order to make + // sure we have a local payload to fall back to in the event of the blinded block + // flow failing. + chain + .prepare_beacon_proposer(current_slot) + .await + .map_err(|e| { + warp_utils::reject::custom_bad_request(format!( + "error updating proposer preparations: {:?}", + e + )) + })?; - let builder = execution_layer - .builder() - .as_ref() - .ok_or(BeaconChainError::BuilderMissing) - .map_err(warp_utils::reject::beacon_chain_error)?; + let builder = execution_layer + .builder() + .as_ref() + .ok_or(BeaconChainError::BuilderMissing) + .map_err(warp_utils::reject::beacon_chain_error)?; - info!( - log, - "Forwarding register validator request to connected builder"; - "count" => filtered_registration_data.len(), - ); + info!( + log, + "Forwarding register validator request to connected builder"; + "count" => filtered_registration_data.len(), + ); - builder - .post_builder_validators(&filtered_registration_data) - .await - .map(|resp| warp::reply::json(&resp).into_response()) - .map_err(|e| { - warn!( - log, - "Relay error when registering validator(s)"; - "num_registrations" => filtered_registration_data.len(), - "error" => ?e - ); - // Forward the HTTP status code if we are able to, otherwise fall back - // to a server error. - if let eth2::Error::ServerMessage(message) = e { - if message.code == StatusCode::BAD_REQUEST.as_u16() { - return warp_utils::reject::custom_bad_request(message.message); - } else { - // According to the spec this response should only be a 400 or 500, - // so we fall back to a 500 here. - return warp_utils::reject::custom_server_error(message.message); + builder + .post_builder_validators(&filtered_registration_data) + .await + .map(|resp| warp::reply::json(&resp).into_response()) + .map_err(|e| { + warn!( + log, + "Relay error when registering validator(s)"; + "num_registrations" => filtered_registration_data.len(), + "error" => ?e + ); + // Forward the HTTP status code if we are able to, otherwise fall back + // to a server error. + if let eth2::Error::ServerMessage(message) = e { + if message.code == StatusCode::BAD_REQUEST.as_u16() { + return warp_utils::reject::custom_bad_request(message.message); + } else { + // According to the spec this response should only be a 400 or 500, + // so we fall back to a 500 here. + return warp_utils::reject::custom_server_error( + message.message, + ); + } } - } - warp_utils::reject::custom_server_error(format!("{e:?}")) - }) + warp_utils::reject::custom_server_error(format!("{e:?}")) + }) + }) }, ); // POST validator/sync_committee_subscriptions @@ -3318,15 +3460,17 @@ pub fn serve( .and(warp::path::end()) .and(warp::body::json()) .and(validator_subscription_tx_filter) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) .and(log_filter.clone()) .and_then( |subscriptions: Vec, validator_subscription_tx: Sender, + task_spawner: TaskSpawner, chain: Arc>, log: Logger | { - blocking_json_task(move || { + task_spawner.blocking_json_task(Priority::P0, move || { for subscription in subscriptions { chain .validator_monitor @@ -3359,10 +3503,13 @@ pub fn serve( .and(warp::path("liveness")) .and(warp::path::end()) .and(warp::body::json()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) .and_then( - |request_data: api_types::LivenessRequestData, chain: Arc>| { - blocking_json_task(move || { + |request_data: api_types::LivenessRequestData, + task_spawner: TaskSpawner, + chain: Arc>| { + task_spawner.blocking_json_task(Priority::P0, move || { // Ensure the request is for either the current, previous or next epoch. let current_epoch = chain .epoch() @@ -3401,8 +3548,9 @@ pub fn serve( let get_lighthouse_health = warp::path("lighthouse") .and(warp::path("health")) .and(warp::path::end()) - .and_then(|| { - blocking_json_task(move || { + .and(task_spawner_filter.clone()) + .and_then(|task_spawner: TaskSpawner| { + task_spawner.blocking_json_task(Priority::P1, move || { eth2::lighthouse::Health::observe() .map(api_types::GenericResponse::from) .map_err(warp_utils::reject::custom_bad_request) @@ -3414,13 +3562,18 @@ pub fn serve( .and(warp::path("ui")) .and(warp::path("health")) .and(warp::path::end()) + .and(task_spawner_filter.clone()) .and(system_info_filter) .and(app_start_filter) .and(data_dir_filter) .and(network_globals.clone()) .and_then( - |sysinfo, app_start: std::time::Instant, data_dir, network_globals| { - blocking_json_task(move || { + |task_spawner: TaskSpawner, + sysinfo, + app_start: std::time::Instant, + data_dir, + network_globals| { + task_spawner.blocking_json_task(Priority::P1, move || { let app_uptime = app_start.elapsed().as_secs(); Ok(api_types::GenericResponse::from(observe_system_health_bn( sysinfo, @@ -3437,12 +3590,15 @@ pub fn serve( .and(warp::path("ui")) .and(warp::path("validator_count")) .and(warp::path::end()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) - .and_then(|chain: Arc>| { - blocking_json_task(move || { - ui::get_validator_count(chain).map(api_types::GenericResponse::from) - }) - }); + .and_then( + |task_spawner: TaskSpawner, chain: Arc>| { + task_spawner.blocking_json_task(Priority::P1, move || { + ui::get_validator_count(chain).map(api_types::GenericResponse::from) + }) + }, + ); // POST lighthouse/ui/validator_metrics let post_lighthouse_ui_validator_metrics = warp::path("lighthouse") @@ -3450,10 +3606,13 @@ pub fn serve( .and(warp::path("validator_metrics")) .and(warp::path::end()) .and(warp::body::json()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) .and_then( - |request_data: ui::ValidatorMetricsRequestData, chain: Arc>| { - blocking_json_task(move || { + |request_data: ui::ValidatorMetricsRequestData, + task_spawner: TaskSpawner, + chain: Arc>| { + task_spawner.blocking_json_task(Priority::P1, move || { ui::post_validator_monitor_metrics(request_data, chain) .map(api_types::GenericResponse::from) }) @@ -3466,10 +3625,13 @@ pub fn serve( .and(warp::path("validator_info")) .and(warp::path::end()) .and(warp::body::json()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) .and_then( - |request_data: ui::ValidatorInfoRequestData, chain: Arc>| { - blocking_json_task(move || { + |request_data: ui::ValidatorInfoRequestData, + task_spawner: TaskSpawner, + chain: Arc>| { + task_spawner.blocking_json_task(Priority::P1, move || { ui::get_validator_info(request_data, chain) .map(api_types::GenericResponse::from) }) @@ -3480,21 +3642,26 @@ pub fn serve( let get_lighthouse_syncing = warp::path("lighthouse") .and(warp::path("syncing")) .and(warp::path::end()) + .and(task_spawner_filter.clone()) .and(network_globals.clone()) - .and_then(|network_globals: Arc>| { - blocking_json_task(move || { - Ok(api_types::GenericResponse::from( - network_globals.sync_state(), - )) - }) - }); + .and_then( + |task_spawner: TaskSpawner, + network_globals: Arc>| { + task_spawner.blocking_json_task(Priority::P1, move || { + Ok(api_types::GenericResponse::from( + network_globals.sync_state(), + )) + }) + }, + ); // GET lighthouse/nat let get_lighthouse_nat = warp::path("lighthouse") .and(warp::path("nat")) + .and(task_spawner_filter.clone()) .and(warp::path::end()) - .and_then(|| { - blocking_json_task(move || { + .and_then(|task_spawner: TaskSpawner| { + task_spawner.blocking_json_task(Priority::P1, move || { Ok(api_types::GenericResponse::from( lighthouse_network::metrics::NAT_OPEN .as_ref() @@ -3509,57 +3676,70 @@ pub fn serve( let get_lighthouse_peers = warp::path("lighthouse") .and(warp::path("peers")) .and(warp::path::end()) + .and(task_spawner_filter.clone()) .and(network_globals.clone()) - .and_then(|network_globals: Arc>| { - blocking_json_task(move || { - Ok(network_globals - .peers - .read() - .peers() - .map(|(peer_id, peer_info)| eth2::lighthouse::Peer { - peer_id: peer_id.to_string(), - peer_info: peer_info.clone(), - }) - .collect::>()) - }) - }); + .and_then( + |task_spawner: TaskSpawner, + network_globals: Arc>| { + task_spawner.blocking_json_task(Priority::P1, move || { + Ok(network_globals + .peers + .read() + .peers() + .map(|(peer_id, peer_info)| eth2::lighthouse::Peer { + peer_id: peer_id.to_string(), + peer_info: peer_info.clone(), + }) + .collect::>()) + }) + }, + ); // GET lighthouse/peers/connected let get_lighthouse_peers_connected = warp::path("lighthouse") .and(warp::path("peers")) .and(warp::path("connected")) .and(warp::path::end()) + .and(task_spawner_filter.clone()) .and(network_globals) - .and_then(|network_globals: Arc>| { - blocking_json_task(move || { - Ok(network_globals - .peers - .read() - .connected_peers() - .map(|(peer_id, peer_info)| eth2::lighthouse::Peer { - peer_id: peer_id.to_string(), - peer_info: peer_info.clone(), - }) - .collect::>()) - }) - }); + .and_then( + |task_spawner: TaskSpawner, + network_globals: Arc>| { + task_spawner.blocking_json_task(Priority::P1, move || { + Ok(network_globals + .peers + .read() + .connected_peers() + .map(|(peer_id, peer_info)| eth2::lighthouse::Peer { + peer_id: peer_id.to_string(), + peer_info: peer_info.clone(), + }) + .collect::>()) + }) + }, + ); // GET lighthouse/proto_array let get_lighthouse_proto_array = warp::path("lighthouse") .and(warp::path("proto_array")) .and(warp::path::end()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) - .and_then(|chain: Arc>| { - blocking_response_task(move || { - Ok::<_, warp::Rejection>(warp::reply::json(&api_types::GenericResponseRef::from( - chain - .canonical_head - .fork_choice_read_lock() - .proto_array() - .core_proto_array(), - ))) - }) - }); + .and_then( + |task_spawner: TaskSpawner, chain: Arc>| { + task_spawner.blocking_response_task(Priority::P1, move || { + Ok::<_, warp::Rejection>(warp::reply::json( + &api_types::GenericResponseRef::from( + chain + .canonical_head + .fork_choice_read_lock() + .proto_array() + .core_proto_array(), + ), + )) + }) + }, + ); // GET lighthouse/validator_inclusion/{epoch}/{validator_id} let get_lighthouse_validator_inclusion_global = warp::path("lighthouse") @@ -3567,10 +3747,14 @@ pub fn serve( .and(warp::path::param::()) .and(warp::path::param::()) .and(warp::path::end()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) .and_then( - |epoch: Epoch, validator_id: ValidatorId, chain: Arc>| { - blocking_json_task(move || { + |epoch: Epoch, + validator_id: ValidatorId, + task_spawner: TaskSpawner, + chain: Arc>| { + task_spawner.blocking_json_task(Priority::P1, move || { validator_inclusion::validator_inclusion_data(epoch, &validator_id, &chain) .map(api_types::GenericResponse::from) }) @@ -3583,82 +3767,94 @@ pub fn serve( .and(warp::path::param::()) .and(warp::path("global")) .and(warp::path::end()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) - .and_then(|epoch: Epoch, chain: Arc>| { - blocking_json_task(move || { - validator_inclusion::global_validator_inclusion_data(epoch, &chain) - .map(api_types::GenericResponse::from) - }) - }); + .and_then( + |epoch: Epoch, task_spawner: TaskSpawner, chain: Arc>| { + task_spawner.blocking_json_task(Priority::P1, move || { + validator_inclusion::global_validator_inclusion_data(epoch, &chain) + .map(api_types::GenericResponse::from) + }) + }, + ); // GET lighthouse/eth1/syncing let get_lighthouse_eth1_syncing = warp::path("lighthouse") .and(warp::path("eth1")) .and(warp::path("syncing")) .and(warp::path::end()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) - .and_then(|chain: Arc>| { - blocking_json_task(move || { - let current_slot_opt = chain.slot().ok(); + .and_then( + |task_spawner: TaskSpawner, chain: Arc>| { + task_spawner.blocking_json_task(Priority::P1, move || { + let current_slot_opt = chain.slot().ok(); - chain - .eth1_chain - .as_ref() - .ok_or_else(|| { - warp_utils::reject::custom_not_found( - "Eth1 sync is disabled. See the --eth1 CLI flag.".to_string(), - ) - }) - .and_then(|eth1| { - eth1.sync_status(chain.genesis_time, current_slot_opt, &chain.spec) - .ok_or_else(|| { - warp_utils::reject::custom_server_error( - "Unable to determine Eth1 sync status".to_string(), - ) - }) - }) - .map(api_types::GenericResponse::from) - }) - }); + chain + .eth1_chain + .as_ref() + .ok_or_else(|| { + warp_utils::reject::custom_not_found( + "Eth1 sync is disabled. See the --eth1 CLI flag.".to_string(), + ) + }) + .and_then(|eth1| { + eth1.sync_status(chain.genesis_time, current_slot_opt, &chain.spec) + .ok_or_else(|| { + warp_utils::reject::custom_server_error( + "Unable to determine Eth1 sync status".to_string(), + ) + }) + }) + .map(api_types::GenericResponse::from) + }) + }, + ); // GET lighthouse/eth1/block_cache let get_lighthouse_eth1_block_cache = warp::path("lighthouse") .and(warp::path("eth1")) .and(warp::path("block_cache")) .and(warp::path::end()) + .and(task_spawner_filter.clone()) .and(eth1_service_filter.clone()) - .and_then(|eth1_service: eth1::Service| { - blocking_json_task(move || { - Ok(api_types::GenericResponse::from( - eth1_service - .blocks() - .read() - .iter() - .cloned() - .collect::>(), - )) - }) - }); + .and_then( + |task_spawner: TaskSpawner, eth1_service: eth1::Service| { + task_spawner.blocking_json_task(Priority::P1, move || { + Ok(api_types::GenericResponse::from( + eth1_service + .blocks() + .read() + .iter() + .cloned() + .collect::>(), + )) + }) + }, + ); // GET lighthouse/eth1/deposit_cache let get_lighthouse_eth1_deposit_cache = warp::path("lighthouse") .and(warp::path("eth1")) .and(warp::path("deposit_cache")) .and(warp::path::end()) + .and(task_spawner_filter.clone()) .and(eth1_service_filter) - .and_then(|eth1_service: eth1::Service| { - blocking_json_task(move || { - Ok(api_types::GenericResponse::from( - eth1_service - .deposits() - .read() - .cache - .iter() - .cloned() - .collect::>(), - )) - }) - }); + .and_then( + |task_spawner: TaskSpawner, eth1_service: eth1::Service| { + task_spawner.blocking_json_task(Priority::P1, move || { + Ok(api_types::GenericResponse::from( + eth1_service + .deposits() + .read() + .cache + .iter() + .cloned() + .collect::>(), + )) + }) + }, + ); // GET lighthouse/beacon/states/{state_id}/ssz let get_lighthouse_beacon_states_ssz = warp::path("lighthouse") @@ -3667,42 +3863,50 @@ pub fn serve( .and(warp::path::param::()) .and(warp::path("ssz")) .and(warp::path::end()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) - .and_then(|state_id: StateId, chain: Arc>| { - blocking_response_task(move || { - // This debug endpoint provides no indication of optimistic status. - let (state, _execution_optimistic, _finalized) = state_id.state(&chain)?; - Response::builder() - .status(200) - .header("Content-Type", "application/ssz") - .body(state.as_ssz_bytes()) - .map_err(|e| { - warp_utils::reject::custom_server_error(format!( - "failed to create response: {}", - e - )) - }) - }) - }); + .and_then( + |state_id: StateId, + task_spawner: TaskSpawner, + chain: Arc>| { + task_spawner.blocking_response_task(Priority::P1, move || { + // This debug endpoint provides no indication of optimistic status. + let (state, _execution_optimistic, _finalized) = state_id.state(&chain)?; + Response::builder() + .status(200) + .header("Content-Type", "application/ssz") + .body(state.as_ssz_bytes()) + .map_err(|e| { + warp_utils::reject::custom_server_error(format!( + "failed to create response: {}", + e + )) + }) + }) + }, + ); // GET lighthouse/staking let get_lighthouse_staking = warp::path("lighthouse") .and(warp::path("staking")) .and(warp::path::end()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) - .and_then(|chain: Arc>| { - blocking_json_task(move || { - if chain.eth1_chain.is_some() { - Ok(()) - } else { - Err(warp_utils::reject::custom_not_found( - "staking is not enabled, \ + .and_then( + |task_spawner: TaskSpawner, chain: Arc>| { + task_spawner.blocking_json_task(Priority::P1, move || { + if chain.eth1_chain.is_some() { + Ok(()) + } else { + Err(warp_utils::reject::custom_not_found( + "staking is not enabled, \ see the --staking CLI flag" - .to_string(), - )) - } - }) - }); + .to_string(), + )) + } + }) + }, + ); let database_path = warp::path("lighthouse").and(warp::path("database")); @@ -3710,31 +3914,41 @@ pub fn serve( let get_lighthouse_database_info = database_path .and(warp::path("info")) .and(warp::path::end()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) - .and_then(|chain: Arc>| blocking_json_task(move || database::info(chain))); + .and_then( + |task_spawner: TaskSpawner, chain: Arc>| { + task_spawner.blocking_json_task(Priority::P1, move || database::info(chain)) + }, + ); // POST lighthouse/database/reconstruct let post_lighthouse_database_reconstruct = database_path .and(warp::path("reconstruct")) .and(warp::path::end()) .and(not_while_syncing_filter) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) - .and_then(|chain: Arc>| { - blocking_json_task(move || { - chain.store_migrator.process_reconstruction(); - Ok("success") - }) - }); + .and_then( + |task_spawner: TaskSpawner, chain: Arc>| { + task_spawner.blocking_json_task(Priority::P1, move || { + chain.store_migrator.process_reconstruction(); + Ok("success") + }) + }, + ); // POST lighthouse/database/historical_blocks let post_lighthouse_database_historical_blocks = database_path .and(warp::path("historical_blocks")) .and(warp::path::end()) .and(warp::body::json()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) .and(log_filter.clone()) .and_then( |blocks: Vec>>, + task_spawner: TaskSpawner, chain: Arc>, log: Logger| { info!( @@ -3743,7 +3957,9 @@ pub fn serve( "count" => blocks.len(), "source" => "http_api" ); - blocking_json_task(move || database::historical_blocks(chain, blocks)) + task_spawner.blocking_json_task(Priority::P1, move || { + database::historical_blocks(chain, blocks) + }) }, ); @@ -3753,10 +3969,13 @@ pub fn serve( .and(warp::path("block_rewards")) .and(warp::query::()) .and(warp::path::end()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) .and(log_filter.clone()) - .and_then(|query, chain, log| { - blocking_json_task(move || block_rewards::get_block_rewards(query, chain, log)) + .and_then(|query, task_spawner: TaskSpawner, chain, log| { + task_spawner.blocking_json_task(Priority::P1, move || { + block_rewards::get_block_rewards(query, chain, log) + }) }); // POST lighthouse/analysis/block_rewards @@ -3765,11 +3984,16 @@ pub fn serve( .and(warp::path("block_rewards")) .and(warp::body::json()) .and(warp::path::end()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) .and(log_filter.clone()) - .and_then(|blocks, chain, log| { - blocking_json_task(move || block_rewards::compute_block_rewards(blocks, chain, log)) - }); + .and_then( + |blocks, task_spawner: TaskSpawner, chain, log| { + task_spawner.blocking_json_task(Priority::P1, move || { + block_rewards::compute_block_rewards(blocks, chain, log) + }) + }, + ); // GET lighthouse/analysis/attestation_performance/{index} let get_lighthouse_attestation_performance = warp::path("lighthouse") @@ -3778,12 +4002,15 @@ pub fn serve( .and(warp::path::param::()) .and(warp::query::()) .and(warp::path::end()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) - .and_then(|target, query, chain: Arc>| { - blocking_json_task(move || { - attestation_performance::get_attestation_performance(target, query, chain) - }) - }); + .and_then( + |target, query, task_spawner: TaskSpawner, chain: Arc>| { + task_spawner.blocking_json_task(Priority::P1, move || { + attestation_performance::get_attestation_performance(target, query, chain) + }) + }, + ); // GET lighthouse/analysis/block_packing_efficiency let get_lighthouse_block_packing_efficiency = warp::path("lighthouse") @@ -3791,35 +4018,45 @@ pub fn serve( .and(warp::path("block_packing_efficiency")) .and(warp::query::()) .and(warp::path::end()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) - .and_then(|query, chain: Arc>| { - blocking_json_task(move || { - block_packing_efficiency::get_block_packing_efficiency(query, chain) - }) - }); + .and_then( + |query, task_spawner: TaskSpawner, chain: Arc>| { + task_spawner.blocking_json_task(Priority::P1, move || { + block_packing_efficiency::get_block_packing_efficiency(query, chain) + }) + }, + ); // GET lighthouse/merge_readiness let get_lighthouse_merge_readiness = warp::path("lighthouse") .and(warp::path("merge_readiness")) .and(warp::path::end()) + .and(task_spawner_filter.clone()) .and(chain_filter.clone()) - .and_then(|chain: Arc>| async move { - let merge_readiness = chain.check_merge_readiness().await; - Ok::<_, warp::reject::Rejection>( - warp::reply::json(&api_types::GenericResponse::from(merge_readiness)) - .into_response(), - ) - }); + .and_then( + |task_spawner: TaskSpawner, chain: Arc>| { + task_spawner.spawn_async_with_rejection(Priority::P1, async move { + let merge_readiness = chain.check_merge_readiness().await; + Ok::<_, warp::reject::Rejection>( + warp::reply::json(&api_types::GenericResponse::from(merge_readiness)) + .into_response(), + ) + }) + }, + ); let get_events = eth_v1 .and(warp::path("events")) .and(warp::path::end()) .and(multi_key_query::()) + .and(task_spawner_filter.clone()) .and(chain_filter) .and_then( |topics_res: Result, + task_spawner: TaskSpawner, chain: Arc>| { - blocking_response_task(move || { + task_spawner.blocking_response_task(Priority::P0, move || { let topics = topics_res?; // for each topic subscribed spawn a new subscription let mut receivers = Vec::with_capacity(topics.topics.len()); @@ -3890,38 +4127,46 @@ pub fn serve( let lighthouse_log_events = warp::path("lighthouse") .and(warp::path("logs")) .and(warp::path::end()) + .and(task_spawner_filter) .and(sse_component_filter) - .and_then(|sse_component: Option| { - blocking_response_task(move || { - if let Some(logging_components) = sse_component { - // Build a JSON stream - let s = - BroadcastStream::new(logging_components.sender.subscribe()).map(|msg| { - match msg { - Ok(data) => { - // Serialize to json - match data.to_json_string() { - // Send the json as a Server Side Event - Ok(json) => Ok(Event::default().data(json)), - Err(e) => Err(warp_utils::reject::server_sent_event_error( - format!("Unable to serialize to JSON {}", e), - )), + .and_then( + |task_spawner: TaskSpawner, sse_component: Option| { + task_spawner.blocking_response_task(Priority::P1, move || { + if let Some(logging_components) = sse_component { + // Build a JSON stream + let s = BroadcastStream::new(logging_components.sender.subscribe()).map( + |msg| { + match msg { + Ok(data) => { + // Serialize to json + match data.to_json_string() { + // Send the json as a Server Side Event + Ok(json) => Ok(Event::default().data(json)), + Err(e) => { + Err(warp_utils::reject::server_sent_event_error( + format!("Unable to serialize to JSON {}", e), + )) + } + } } + Err(e) => Err(warp_utils::reject::server_sent_event_error( + format!("Unable to receive event {}", e), + )), } - Err(e) => Err(warp_utils::reject::server_sent_event_error( - format!("Unable to receive event {}", e), - )), - } - }); + }, + ); - Ok::<_, warp::Rejection>(warp::sse::reply(warp::sse::keep_alive().stream(s))) - } else { - Err(warp_utils::reject::custom_server_error( - "SSE Logging is not enabled".to_string(), - )) - } - }) - }); + Ok::<_, warp::Rejection>(warp::sse::reply( + warp::sse::keep_alive().stream(s), + )) + } else { + Err(warp_utils::reject::custom_server_error( + "SSE Logging is not enabled".to_string(), + )) + } + }) + }, + ); // Define the ultimate set of routes that will be provided to the server. // Use `uor` rather than `or` in order to simplify types (see `UnifyingOrFilter`). diff --git a/beacon_node/http_api/src/task_spawner.rs b/beacon_node/http_api/src/task_spawner.rs index 0f6717f3a2..0a7a79cf0a 100644 --- a/beacon_node/http_api/src/task_spawner.rs +++ b/beacon_node/http_api/src/task_spawner.rs @@ -71,14 +71,66 @@ impl TaskSpawner { } } - pub async fn async_task( - &self, + pub async fn blocking_response_task( + self, priority: Priority, - func: impl Future> + Send + Sync + 'static, + func: F, ) -> Result where - T: Serialize + Send + 'static, + F: FnOnce() -> Result + Send + Sync + 'static, + T: Reply + Send + 'static, { + if let Some(beacon_processor_send) = &self.beacon_processor_send { + // Create a closure that will execute `func` and send the result to + // a channel held by this thread. + let (tx, rx) = oneshot::channel(); + let process_fn = move || { + // Execute the function, collect the return value. + let func_result = func(); + // Send the result down the channel. Ignore any failures; the + // send can only fail if the receiver is dropped. + let _ = tx.send(func_result); + }; + let process_fn = BlockingOrAsync::Blocking(Box::new(process_fn)); + + let error_message = + match beacon_processor_send.try_send(priority.work_event(process_fn)) { + Ok(()) => { + match rx.await { + // The beacon processor executed the task and sent a result. + Ok(func_result) => { + return func_result.map(Reply::into_response) + } + // The beacon processor dropped the channel without sending a + // result. The beacon processor dropped this task because its + // queues are full or it's shutting down. + Err(_) => "The task did not execute. The server is overloaded or shutting down.", + } + } + Err(TrySendError::Full(_)) => "The task was dropped. The server is overloaded.", + Err(TrySendError::Closed(_)) => { + "The task was dropped. The server is shutting down." + } + }; + + let error_response = warp::reply::with_status( + warp::reply::json(&error_message), + eth2::StatusCode::INTERNAL_SERVER_ERROR, + ) + .into_response(); + Ok(error_response) + } else { + // There is no beacon processor so spawn a task directly on the + // tokio executor. + warp_utils::task::blocking_response_task(func).await + } + } + + pub async fn spawn_async_with_rejection( + self, + priority: Priority, + func: impl Future> + Send + Sync + 'static, + ) -> Result { if let Some(beacon_processor_send) = &self.beacon_processor_send { // Create a wrapper future that will execute `func` and send the // result to a channel held by this thread. @@ -90,29 +142,103 @@ impl TaskSpawner { // send can only fail if the receiver is dropped. let _ = tx.send(func_result); }; + let process_fn = BlockingOrAsync::Async(Box::pin(process_fn)); // Send the function to the beacon processor for execution at some arbitrary time. - send_to_beacon_processor( - beacon_processor_send, - priority, - BlockingOrAsync::Async(Box::pin(process_fn)), - rx, + let error_message = + match beacon_processor_send.try_send(priority.work_event(process_fn)) { + Ok(()) => { + match rx.await { + // The beacon processor executed the task and sent a result. + Ok(func_result) => { + return func_result + } + // The beacon processor dropped the channel without sending a + // result. The beacon processor dropped this task because its + // queues are full or it's shutting down. + Err(_) => "The task did not execute. The server is overloaded or shutting down.", + } + } + Err(TrySendError::Full(_)) => "The task was dropped. The server is overloaded.", + Err(TrySendError::Closed(_)) => { + "The task was dropped. The server is shutting down." + } + }; + + let error_response = warp::reply::with_status( + warp::reply::json(&error_message), + eth2::StatusCode::INTERNAL_SERVER_ERROR, ) - .await + .into_response(); + Ok(error_response) } else { // There is no beacon processor so spawn a task directly on the // tokio executor. - tokio::task::spawn(func) - .await - .map(|r| r.map(|r| warp::reply::json(&r).into_response())) - .unwrap_or_else(|e| { - let response = warp::reply::with_status( - warp::reply::json(&format!("Tokio did not execute task: {e:?}")), - eth2::StatusCode::INTERNAL_SERVER_ERROR, - ) - .into_response(); - Ok(response) - }) + tokio::task::spawn(func).await.unwrap_or_else(|e| { + let response = warp::reply::with_status( + warp::reply::json(&format!("Tokio did not execute task: {e:?}")), + eth2::StatusCode::INTERNAL_SERVER_ERROR, + ) + .into_response(); + Ok(response) + }) + } + } + + pub async fn spawn_async( + self, + priority: Priority, + func: impl Future + Send + Sync + 'static, + ) -> Response { + if let Some(beacon_processor_send) = &self.beacon_processor_send { + // Create a wrapper future that will execute `func` and send the + // result to a channel held by this thread. + let (tx, rx) = oneshot::channel(); + let process_fn = async move { + // Await the future, collect the return value. + let func_result = func.await; + // Send the result down the channel. Ignore any failures; the + // send can only fail if the receiver is dropped. + let _ = tx.send(func_result); + }; + let process_fn = BlockingOrAsync::Async(Box::pin(process_fn)); + + // Send the function to the beacon processor for execution at some arbitrary time. + let error_message = + match beacon_processor_send.try_send(priority.work_event(process_fn)) { + Ok(()) => { + match rx.await { + // The beacon processor executed the task and sent a result. + Ok(func_result) => { + return func_result + } + // The beacon processor dropped the channel without sending a + // result. The beacon processor dropped this task because its + // queues are full or it's shutting down. + Err(_) => "The task did not execute. The server is overloaded or shutting down.", + } + } + Err(TrySendError::Full(_)) => "The task was dropped. The server is overloaded.", + Err(TrySendError::Closed(_)) => { + "The task was dropped. The server is shutting down." + } + }; + + warp::reply::with_status( + warp::reply::json(&error_message), + eth2::StatusCode::INTERNAL_SERVER_ERROR, + ) + .into_response() + } else { + // There is no beacon processor so spawn a task directly on the + // tokio executor. + tokio::task::spawn(func).await.unwrap_or_else(|e| { + warp::reply::with_status( + warp::reply::json(&format!("Tokio did not execute task: {e:?}")), + eth2::StatusCode::INTERNAL_SERVER_ERROR, + ) + .into_response() + }) } } }