cerc-io/lighthouse
16
0
Fork 0
Code Issues 1 Pull Requests Packages Projects Releases Wiki Activity

Improve HTTP API error messages + tweaks (#4595)

Browse Source 
## Issue Addressed

Closes #3404 (mostly)

## Proposed Changes

- Remove all uses of Warp's `and_then` (which backtracks) in favour of `then` (which doesn't).
- Bump the priority of the `POST` method for `v2/blocks` to `P0`. Publishing a block needs to happen quickly.
- Run the new SSZ POST endpoints on the beacon processor. I think this was missed in between merging #4462 and #4504/#4479.
- Fix a minor issue in the validator registrations endpoint whereby an error from spawning the task on the beacon processor would be dropped.

## Additional Info

I've tested this manually and can confirm that we no longer get the dreaded `Unsupported endpoint version` errors for queries like:

```
$ curl -X POST -H "Content-Type: application/json" --data @block.json "http://localhost:5052/eth/v2/beacon/blocks" | jq
{
  "code": 400,
  "message": "BAD_REQUEST: WeakSubjectivityConflict",
  "stacktraces": []
}
```

```
$ curl -X POST -H "Content-Type: application/octet-stream" --data @block.json "http://localhost:5052/eth/v2/beacon/blocks" | jq
{
  "code": 400,
  "message": "BAD_REQUEST: invalid SSZ: OffsetOutOfBounds(572530811)",
  "stacktraces": []
}
```

```
$ curl "http://localhost:5052/eth/v2/validator/blocks/7067595"
{"code":400,"message":"BAD_REQUEST: invalid query: Invalid query string","stacktraces":[]}
```

However, I can still trigger it by leaving off the `Content-Type`. We can re-test this aspect with #4575.
This commit is contained in:
Michael Sproul 2023-08-14 04:06:37 +00:00
parent 912f869829
commit 249f85f1d9
2 changed files with 227 additions and 227 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

362
beacon_node/http_api/src/lib.rs
View File

@ -516,7 +516,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::end()) .and(warp::path::end())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| { |task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| {
task_spawner.blocking_json_task(Priority::P1, move || { task_spawner.blocking_json_task(Priority::P1, move || {
let genesis_data = api_types::GenesisData { let genesis_data = api_types::GenesisData {
@ -549,7 +549,7 @@ pub fn serve<T: BeaconChainTypes>(
.clone() .clone()
.and(warp::path("root")) .and(warp::path("root"))
.and(warp::path::end()) .and(warp::path::end())
.and_then( .then(
|state_id: StateId, |state_id: StateId,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>| { chain: Arc<BeaconChain<T>>| {
@ -570,7 +570,7 @@ pub fn serve<T: BeaconChainTypes>(
.clone() .clone()
.and(warp::path("fork")) .and(warp::path("fork"))
.and(warp::path::end()) .and(warp::path::end())
.and_then( .then(
|state_id: StateId, |state_id: StateId,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>| { chain: Arc<BeaconChain<T>>| {
@ -591,7 +591,7 @@ pub fn serve<T: BeaconChainTypes>(
.clone() .clone()
.and(warp::path("finality_checkpoints")) .and(warp::path("finality_checkpoints"))
.and(warp::path::end()) .and(warp::path::end())
.and_then( .then(
|state_id: StateId, |state_id: StateId,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>| { chain: Arc<BeaconChain<T>>| {
@ -627,7 +627,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path("validator_balances")) .and(warp::path("validator_balances"))
.and(warp::path::end()) .and(warp::path::end())
.and(multi_key_query::<api_types::ValidatorBalancesQuery>()) .and(multi_key_query::<api_types::ValidatorBalancesQuery>())
.and_then( .then(
|state_id: StateId, |state_id: StateId,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>, chain: Arc<BeaconChain<T>>,
@ -685,7 +685,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path("validators")) .and(warp::path("validators"))
.and(warp::path::end()) .and(warp::path::end())
.and(multi_key_query::<api_types::ValidatorsQuery>()) .and(multi_key_query::<api_types::ValidatorsQuery>())
.and_then( .then(
|state_id: StateId, |state_id: StateId,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>, chain: Arc<BeaconChain<T>>,
@ -769,7 +769,7 @@ pub fn serve<T: BeaconChainTypes>(
)) ))
})) }))
.and(warp::path::end()) .and(warp::path::end())
.and_then( .then(
|state_id: StateId, |state_id: StateId,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>, chain: Arc<BeaconChain<T>>,
@ -837,7 +837,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path("committees")) .and(warp::path("committees"))
.and(warp::query::<api_types::CommitteesQuery>()) .and(warp::query::<api_types::CommitteesQuery>())
.and(warp::path::end()) .and(warp::path::end())
.and_then( .then(
|state_id: StateId, |state_id: StateId,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>, chain: Arc<BeaconChain<T>>,
@ -1020,7 +1020,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path("sync_committees")) .and(warp::path("sync_committees"))
.and(warp::query::<api_types::SyncCommitteesQuery>()) .and(warp::query::<api_types::SyncCommitteesQuery>())
.and(warp::path::end()) .and(warp::path::end())
.and_then( .then(
|state_id: StateId, |state_id: StateId,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>, chain: Arc<BeaconChain<T>>,
@ -1086,7 +1086,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path("randao")) .and(warp::path("randao"))
.and(warp::query::<api_types::RandaoQuery>()) .and(warp::query::<api_types::RandaoQuery>())
.and(warp::path::end()) .and(warp::path::end())
.and_then( .then(
|state_id: StateId, |state_id: StateId,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>, chain: Arc<BeaconChain<T>>,
@ -1128,7 +1128,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::end()) .and(warp::path::end())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and_then( .then(
|query: api_types::HeadersQuery, |query: api_types::HeadersQuery,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>| { chain: Arc<BeaconChain<T>>| {
@ -1228,7 +1228,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::end()) .and(warp::path::end())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and_then( .then(
|block_id: BlockId, |block_id: BlockId,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>| { chain: Arc<BeaconChain<T>>| {
@ -1276,7 +1276,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(chain_filter.clone()) .and(chain_filter.clone())
.and(network_tx_filter.clone()) .and(network_tx_filter.clone())
.and(log_filter.clone()) .and(log_filter.clone())
.and_then( .then(
|block: Arc<SignedBeaconBlock<T::EthSpec>>, |block: Arc<SignedBeaconBlock<T::EthSpec>>,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>, chain: Arc<BeaconChain<T>>,
@ -1302,33 +1302,35 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path("blocks")) .and(warp::path("blocks"))
.and(warp::path::end()) .and(warp::path::end())
.and(warp::body::bytes()) .and(warp::body::bytes())
.and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and(network_tx_filter.clone()) .and(network_tx_filter.clone())
.and(log_filter.clone()) .and(log_filter.clone())
.and_then( .then(
|block_bytes: Bytes, |block_bytes: Bytes,
task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>, chain: Arc<BeaconChain<T>>,
network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>, network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>,
log: Logger| async move { log: Logger| {
let block = match SignedBeaconBlock::<T::EthSpec>::from_ssz_bytes( task_spawner.spawn_async_with_rejection(Priority::P0, async move {
&block_bytes, let block =
&chain.spec, SignedBeaconBlock::<T::EthSpec>::from_ssz_bytes(&block_bytes, &chain.spec)
) { .map_err(|e| {
Ok(data) => data, warp_utils::reject::custom_bad_request(format!(
Err(e) => { "invalid SSZ: {e:?}"
return Err(warp_utils::reject::custom_bad_request(format!("{:?}", e))) ))
} })?;
}; publish_blocks::publish_block(
publish_blocks::publish_block( None,
None, ProvenancedBlock::local(Arc::new(block)),
ProvenancedBlock::local(Arc::new(block)), chain,
chain, &network_tx,
&network_tx, log,
log, BroadcastValidation::default(),
BroadcastValidation::default(), )
) .await
.await .map(|()| warp::reply().into_response())
.map(|()| warp::reply().into_response()) })
}, },
); );
@ -1349,8 +1351,8 @@ pub fn serve<T: BeaconChainTypes>(
chain: Arc<BeaconChain<T>>, chain: Arc<BeaconChain<T>>,
network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>, network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>,
log: Logger| { log: Logger| {
task_spawner.spawn_async(Priority::P1, async move { task_spawner.spawn_async_with_rejection(Priority::P0, async move {
match publish_blocks::publish_block( publish_blocks::publish_block(
None, None,
ProvenancedBlock::local(block), ProvenancedBlock::local(block),
chain, chain,
@ -1359,17 +1361,7 @@ pub fn serve<T: BeaconChainTypes>(
validation_level.broadcast_validation, validation_level.broadcast_validation,
) )
.await .await
{ .map(|()| warp::reply().into_response())
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(),
},
}
}) })
}, },
); );
@ -1380,48 +1372,36 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::query::<api_types::BroadcastValidationQuery>()) .and(warp::query::<api_types::BroadcastValidationQuery>())
.and(warp::path::end()) .and(warp::path::end())
.and(warp::body::bytes()) .and(warp::body::bytes())
.and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and(network_tx_filter.clone()) .and(network_tx_filter.clone())
.and(log_filter.clone()) .and(log_filter.clone())
.then( .then(
|validation_level: api_types::BroadcastValidationQuery, |validation_level: api_types::BroadcastValidationQuery,
block_bytes: Bytes, block_bytes: Bytes,
task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>, chain: Arc<BeaconChain<T>>,
network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>, network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>,
log: Logger| async move { log: Logger| {
let block = match SignedBeaconBlock::<T::EthSpec>::from_ssz_bytes( task_spawner.spawn_async_with_rejection(Priority::P0, async move {
&block_bytes, let block =
&chain.spec, SignedBeaconBlock::<T::EthSpec>::from_ssz_bytes(&block_bytes, &chain.spec)
) { .map_err(|e| {
Ok(data) => data, warp_utils::reject::custom_bad_request(format!(
Err(_) => { "invalid SSZ: {e:?}"
return warp::reply::with_status( ))
StatusCode::BAD_REQUEST, })?;
eth2::StatusCode::BAD_REQUEST, publish_blocks::publish_block(
) None,
.into_response(); ProvenancedBlock::local(Arc::new(block)),
} chain,
}; &network_tx,
match publish_blocks::publish_block( log,
None, validation_level.broadcast_validation,
ProvenancedBlock::local(Arc::new(block)), )
chain, .await
&network_tx, .map(|()| warp::reply().into_response())
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(),
},
}
}, },
); );
@ -1439,7 +1419,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(chain_filter.clone()) .and(chain_filter.clone())
.and(network_tx_filter.clone()) .and(network_tx_filter.clone())
.and(log_filter.clone()) .and(log_filter.clone())
.and_then( .then(
|block: SignedBeaconBlock<T::EthSpec, BlindedPayload<_>>, |block: SignedBeaconBlock<T::EthSpec, BlindedPayload<_>>,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>, chain: Arc<BeaconChain<T>>,
@ -1460,33 +1440,29 @@ pub fn serve<T: BeaconChainTypes>(
); );
// POST beacon/blocks // POST beacon/blocks
let post_beacon_blinded_blocks_ssz = let post_beacon_blinded_blocks_ssz = eth_v1
eth_v1 .and(warp::path("beacon"))
.and(warp::path("beacon")) .and(warp::path("blinded_blocks"))
.and(warp::path("blinded_blocks")) .and(warp::path::end())
.and(warp::path::end()) .and(warp::body::bytes())
.and(warp::body::bytes()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and(network_tx_filter.clone()) .and(network_tx_filter.clone())
.and(log_filter.clone()) .and(log_filter.clone())
.and_then( .then(
|block_bytes: Bytes, |block_bytes: Bytes,
chain: Arc<BeaconChain<T>>, task_spawner: TaskSpawner<T::EthSpec>,
network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>, chain: Arc<BeaconChain<T>>,
log: Logger| async move { network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>,
let block = log: Logger| {
match SignedBeaconBlock::<T::EthSpec, BlindedPayload<_>>::from_ssz_bytes( task_spawner.spawn_async_with_rejection(Priority::P0, async move {
&block_bytes, let block = SignedBeaconBlock::<T::EthSpec, BlindedPayload<_>>::from_ssz_bytes(
&chain.spec, &block_bytes,
) { &chain.spec,
Ok(data) => data, )
Err(e) => { .map_err(|e| {
return Err(warp_utils::reject::custom_bad_request(format!( warp_utils::reject::custom_bad_request(format!("invalid SSZ: {e:?}"))
"{:?}", })?;
e
)))
}
};
publish_blocks::publish_blinded_block( publish_blocks::publish_blinded_block(
block, block,
chain, chain,
@ -1496,8 +1472,9 @@ pub fn serve<T: BeaconChainTypes>(
) )
.await .await
.map(|()| warp::reply().into_response()) .map(|()| warp::reply().into_response())
}, })
); },
);
let post_beacon_blinded_blocks_v2 = eth_v2 let post_beacon_blinded_blocks_v2 = eth_v2
.and(warp::path("beacon")) .and(warp::path("beacon"))
@ -1617,7 +1594,7 @@ pub fn serve<T: BeaconChainTypes>(
.clone() .clone()
.and(warp::path::end()) .and(warp::path::end())
.and(warp::header::optional::<api_types::Accept>("accept")) .and(warp::header::optional::<api_types::Accept>("accept"))
.and_then( .then(
|endpoint_version: EndpointVersion, |endpoint_version: EndpointVersion,
block_id: BlockId, block_id: BlockId,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
@ -1660,7 +1637,7 @@ pub fn serve<T: BeaconChainTypes>(
.clone() .clone()
.and(warp::path("root")) .and(warp::path("root"))
.and(warp::path::end()) .and(warp::path::end())
.and_then( .then(
|block_id: BlockId, |block_id: BlockId,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>| { chain: Arc<BeaconChain<T>>| {
@ -1680,7 +1657,7 @@ pub fn serve<T: BeaconChainTypes>(
.clone() .clone()
.and(warp::path("attestations")) .and(warp::path("attestations"))
.and(warp::path::end()) .and(warp::path::end())
.and_then( .then(
|block_id: BlockId, |block_id: BlockId,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>| { chain: Arc<BeaconChain<T>>| {
@ -1704,7 +1681,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(chain_filter.clone()) .and(chain_filter.clone())
.and(warp::path::end()) .and(warp::path::end())
.and(warp::header::optional::<api_types::Accept>("accept")) .and(warp::header::optional::<api_types::Accept>("accept"))
.and_then( .then(
|block_id: BlockId, |block_id: BlockId,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>, chain: Arc<BeaconChain<T>>,
@ -1762,7 +1739,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::body::json()) .and(warp::body::json())
.and(network_tx_filter.clone()) .and(network_tx_filter.clone())
.and(log_filter.clone()) .and(log_filter.clone())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, |task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>, chain: Arc<BeaconChain<T>>,
attestations: Vec<Attestation<T::EthSpec>>, attestations: Vec<Attestation<T::EthSpec>>,
@ -1904,7 +1881,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path("attestations")) .and(warp::path("attestations"))
.and(warp::path::end()) .and(warp::path::end())
.and(warp::query::<api_types::AttestationPoolQuery>()) .and(warp::query::<api_types::AttestationPoolQuery>())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, |task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>, chain: Arc<BeaconChain<T>>,
query: api_types::AttestationPoolQuery| { query: api_types::AttestationPoolQuery| {
@ -1937,7 +1914,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::end()) .and(warp::path::end())
.and(warp::body::json()) .and(warp::body::json())
.and(network_tx_filter.clone()) .and(network_tx_filter.clone())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, |task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>, chain: Arc<BeaconChain<T>>,
slashing: AttesterSlashing<T::EthSpec>, slashing: AttesterSlashing<T::EthSpec>,
@ -1979,7 +1956,7 @@ pub fn serve<T: BeaconChainTypes>(
.clone() .clone()
.and(warp::path("attester_slashings")) .and(warp::path("attester_slashings"))
.and(warp::path::end()) .and(warp::path::end())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| { |task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| {
task_spawner.blocking_json_task(Priority::P1, move || { task_spawner.blocking_json_task(Priority::P1, move || {
let attestations = chain.op_pool.get_all_attester_slashings(); let attestations = chain.op_pool.get_all_attester_slashings();
@ -1995,7 +1972,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::end()) .and(warp::path::end())
.and(warp::body::json()) .and(warp::body::json())
.and(network_tx_filter.clone()) .and(network_tx_filter.clone())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, |task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>, chain: Arc<BeaconChain<T>>,
slashing: ProposerSlashing, slashing: ProposerSlashing,
@ -2037,7 +2014,7 @@ pub fn serve<T: BeaconChainTypes>(
.clone() .clone()
.and(warp::path("proposer_slashings")) .and(warp::path("proposer_slashings"))
.and(warp::path::end()) .and(warp::path::end())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| { |task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| {
task_spawner.blocking_json_task(Priority::P1, move || { task_spawner.blocking_json_task(Priority::P1, move || {
let attestations = chain.op_pool.get_all_proposer_slashings(); let attestations = chain.op_pool.get_all_proposer_slashings();
@ -2053,7 +2030,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::end()) .and(warp::path::end())
.and(warp::body::json()) .and(warp::body::json())
.and(network_tx_filter.clone()) .and(network_tx_filter.clone())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, |task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>, chain: Arc<BeaconChain<T>>,
exit: SignedVoluntaryExit, exit: SignedVoluntaryExit,
@ -2093,7 +2070,7 @@ pub fn serve<T: BeaconChainTypes>(
.clone() .clone()
.and(warp::path("voluntary_exits")) .and(warp::path("voluntary_exits"))
.and(warp::path::end()) .and(warp::path::end())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| { |task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| {
task_spawner.blocking_json_task(Priority::P1, move || { task_spawner.blocking_json_task(Priority::P1, move || {
let attestations = chain.op_pool.get_all_voluntary_exits(); let attestations = chain.op_pool.get_all_voluntary_exits();
@ -2110,7 +2087,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::body::json()) .and(warp::body::json())
.and(network_tx_filter.clone()) .and(network_tx_filter.clone())
.and(log_filter.clone()) .and(log_filter.clone())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, |task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>, chain: Arc<BeaconChain<T>>,
signatures: Vec<SyncCommitteeMessage>, signatures: Vec<SyncCommitteeMessage>,
@ -2130,7 +2107,7 @@ pub fn serve<T: BeaconChainTypes>(
.clone() .clone()
.and(warp::path("bls_to_execution_changes")) .and(warp::path("bls_to_execution_changes"))
.and(warp::path::end()) .and(warp::path::end())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| { |task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| {
task_spawner.blocking_json_task(Priority::P1, move || { task_spawner.blocking_json_task(Priority::P1, move || {
let address_changes = chain.op_pool.get_all_bls_to_execution_changes(); let address_changes = chain.op_pool.get_all_bls_to_execution_changes();
@ -2147,7 +2124,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::body::json()) .and(warp::body::json())
.and(network_tx_filter.clone()) .and(network_tx_filter.clone())
.and(log_filter.clone()) .and(log_filter.clone())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, |task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>, chain: Arc<BeaconChain<T>>,
address_changes: Vec<SignedBlsToExecutionChange>, address_changes: Vec<SignedBlsToExecutionChange>,
@ -2239,7 +2216,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::header::optional::<api_types::Accept>("accept")) .and(warp::header::optional::<api_types::Accept>("accept"))
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(eth1_service_filter.clone()) .and(eth1_service_filter.clone())
.and_then( .then(
|accept_header: Option<api_types::Accept>, |accept_header: Option<api_types::Accept>,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
eth1_service: eth1::Service| { eth1_service: eth1::Service| {
@ -2293,7 +2270,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path("blocks")) .and(warp::path("blocks"))
.and(block_id_or_err) .and(block_id_or_err)
.and(warp::path::end()) .and(warp::path::end())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, |task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>, chain: Arc<BeaconChain<T>>,
block_id: BlockId| { block_id: BlockId| {
@ -2326,7 +2303,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::param::<Epoch>()) .and(warp::path::param::<Epoch>())
.and(warp::path::end()) .and(warp::path::end())
.and(warp::body::json()) .and(warp::body::json())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, |task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>, chain: Arc<BeaconChain<T>>,
epoch: Epoch, epoch: Epoch,
@ -2378,7 +2355,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::end()) .and(warp::path::end())
.and(warp::body::json()) .and(warp::body::json())
.and(log_filter.clone()) .and(log_filter.clone())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, |task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>, chain: Arc<BeaconChain<T>>,
block_id: BlockId, block_id: BlockId,
@ -2411,7 +2388,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::end()) .and(warp::path::end())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| { |task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| {
task_spawner.blocking_json_task(Priority::P1, move || { task_spawner.blocking_json_task(Priority::P1, move || {
let forks = ForkName::list_all() let forks = ForkName::list_all()
@ -2430,7 +2407,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::end()) .and(warp::path::end())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and_then( .then(
move |task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| { move |task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| {
task_spawner.blocking_json_task(Priority::P0, move || { task_spawner.blocking_json_task(Priority::P0, move || {
let config_and_preset = let config_and_preset =
@ -2446,7 +2423,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::end()) .and(warp::path::end())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| { |task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| {
task_spawner.blocking_json_task(Priority::P1, move || { task_spawner.blocking_json_task(Priority::P1, move || {
Ok(api_types::GenericResponse::from( Ok(api_types::GenericResponse::from(
@ -2477,7 +2454,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::header::optional::<api_types::Accept>("accept")) .and(warp::header::optional::<api_types::Accept>("accept"))
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and_then( .then(
|endpoint_version: EndpointVersion, |endpoint_version: EndpointVersion,
state_id: StateId, state_id: StateId,
accept_header: Option<api_types::Accept>, accept_header: Option<api_types::Accept>,
@ -2537,7 +2514,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::end()) .and(warp::path::end())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and_then( .then(
|endpoint_version: EndpointVersion, |endpoint_version: EndpointVersion,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>| { chain: Arc<BeaconChain<T>>| {
@ -2576,7 +2553,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::end()) .and(warp::path::end())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| { |task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| {
task_spawner.blocking_json_task(Priority::P1, move || { task_spawner.blocking_json_task(Priority::P1, move || {
let beacon_fork_choice = chain.canonical_head.fork_choice_read_lock(); let beacon_fork_choice = chain.canonical_head.fork_choice_read_lock();
@ -2631,7 +2608,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::end()) .and(warp::path::end())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(network_globals.clone()) .and(network_globals.clone())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, |task_spawner: TaskSpawner<T::EthSpec>,
network_globals: Arc<NetworkGlobals<T::EthSpec>>| { network_globals: Arc<NetworkGlobals<T::EthSpec>>| {
task_spawner.blocking_json_task(Priority::P1, move || { task_spawner.blocking_json_task(Priority::P1, move || {
@ -2687,7 +2664,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(network_globals.clone()) .and(network_globals.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, |task_spawner: TaskSpawner<T::EthSpec>,
network_globals: Arc<NetworkGlobals<T::EthSpec>>, network_globals: Arc<NetworkGlobals<T::EthSpec>>,
chain: Arc<BeaconChain<T>>| { chain: Arc<BeaconChain<T>>| {
@ -2738,7 +2715,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(network_globals.clone()) .and(network_globals.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, |task_spawner: TaskSpawner<T::EthSpec>,
network_globals: Arc<NetworkGlobals<T::EthSpec>>, network_globals: Arc<NetworkGlobals<T::EthSpec>>,
chain: Arc<BeaconChain<T>>| { chain: Arc<BeaconChain<T>>| {
@ -2786,7 +2763,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::end()) .and(warp::path::end())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(network_globals.clone()) .and(network_globals.clone())
.and_then( .then(
|requested_peer_id: String, |requested_peer_id: String,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
network_globals: Arc<NetworkGlobals<T::EthSpec>>| { network_globals: Arc<NetworkGlobals<T::EthSpec>>| {
@ -2846,7 +2823,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(multi_key_query::<api_types::PeersQuery>()) .and(multi_key_query::<api_types::PeersQuery>())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(network_globals.clone()) .and(network_globals.clone())
.and_then( .then(
|query_res: Result<api_types::PeersQuery, warp::Rejection>, |query_res: Result<api_types::PeersQuery, warp::Rejection>,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
network_globals: Arc<NetworkGlobals<T::EthSpec>>| { network_globals: Arc<NetworkGlobals<T::EthSpec>>| {
@ -2916,7 +2893,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::end()) .and(warp::path::end())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(network_globals.clone()) .and(network_globals.clone())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, |task_spawner: TaskSpawner<T::EthSpec>,
network_globals: Arc<NetworkGlobals<T::EthSpec>>| { network_globals: Arc<NetworkGlobals<T::EthSpec>>| {
task_spawner.blocking_json_task(Priority::P1, move || { task_spawner.blocking_json_task(Priority::P1, move || {
@ -2969,7 +2946,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and(log_filter.clone()) .and(log_filter.clone())
.and_then( .then(
|epoch: Epoch, |epoch: Epoch,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>, chain: Arc<BeaconChain<T>>,
@ -2995,7 +2972,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and(log_filter.clone()) .and(log_filter.clone())
.and_then( .then(
|endpoint_version: EndpointVersion, |endpoint_version: EndpointVersion,
slot: Slot, slot: Slot,
query: api_types::ValidatorBlocksQuery, query: api_types::ValidatorBlocksQuery,
@ -3064,7 +3041,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::query::<api_types::ValidatorBlocksQuery>()) .and(warp::query::<api_types::ValidatorBlocksQuery>())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and_then( .then(
|slot: Slot, |slot: Slot,
query: api_types::ValidatorBlocksQuery, query: api_types::ValidatorBlocksQuery,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
@ -3121,7 +3098,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(not_while_syncing_filter.clone()) .and(not_while_syncing_filter.clone())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and_then( .then(
|query: api_types::ValidatorAttestationDataQuery, |query: api_types::ValidatorAttestationDataQuery,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>| { chain: Arc<BeaconChain<T>>| {
@ -3156,7 +3133,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(not_while_syncing_filter.clone()) .and(not_while_syncing_filter.clone())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and_then( .then(
|query: api_types::ValidatorAggregateAttestationQuery, |query: api_types::ValidatorAggregateAttestationQuery,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>| { chain: Arc<BeaconChain<T>>| {
@ -3197,7 +3174,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::body::json()) .and(warp::body::json())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and_then( .then(
|epoch: Epoch, |epoch: Epoch,
indices: api_types::ValidatorIndexData, indices: api_types::ValidatorIndexData,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
@ -3223,7 +3200,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::body::json()) .and(warp::body::json())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and_then( .then(
|epoch: Epoch, |epoch: Epoch,
indices: api_types::ValidatorIndexData, indices: api_types::ValidatorIndexData,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
@ -3243,7 +3220,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(not_while_syncing_filter.clone()) .and(not_while_syncing_filter.clone())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and_then( .then(
|sync_committee_data: SyncContributionData, |sync_committee_data: SyncContributionData,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>| { chain: Arc<BeaconChain<T>>| {
@ -3277,7 +3254,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::body::json()) .and(warp::body::json())
.and(network_tx_filter.clone()) .and(network_tx_filter.clone())
.and(log_filter.clone()) .and(log_filter.clone())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, |task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>, chain: Arc<BeaconChain<T>>,
aggregates: Vec<SignedAggregateAndProof<T::EthSpec>>, aggregates: Vec<SignedAggregateAndProof<T::EthSpec>>,
@ -3390,7 +3367,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::body::json()) .and(warp::body::json())
.and(network_tx_filter) .and(network_tx_filter)
.and(log_filter.clone()) .and(log_filter.clone())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, |task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>, chain: Arc<BeaconChain<T>>,
contributions: Vec<SignedContributionAndProof<T::EthSpec>>, contributions: Vec<SignedContributionAndProof<T::EthSpec>>,
@ -3418,7 +3395,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and(log_filter.clone()) .and(log_filter.clone())
.and_then( .then(
|subscriptions: Vec<api_types::BeaconCommitteeSubscription>, |subscriptions: Vec<api_types::BeaconCommitteeSubscription>,
validator_subscription_tx: Sender<ValidatorSubscriptionMessage>, validator_subscription_tx: Sender<ValidatorSubscriptionMessage>,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
@ -3470,7 +3447,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(chain_filter.clone()) .and(chain_filter.clone())
.and(log_filter.clone()) .and(log_filter.clone())
.and(warp::body::json()) .and(warp::body::json())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, |task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>, chain: Arc<BeaconChain<T>>,
log: Logger, log: Logger,
@ -3521,15 +3498,15 @@ pub fn serve<T: BeaconChainTypes>(
.and(chain_filter.clone()) .and(chain_filter.clone())
.and(log_filter.clone()) .and(log_filter.clone())
.and(warp::body::json()) .and(warp::body::json())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, |task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>, chain: Arc<BeaconChain<T>>,
log: Logger, log: Logger,
register_val_data: Vec<SignedValidatorRegistrationData>| async { register_val_data: Vec<SignedValidatorRegistrationData>| async {
let (tx, rx) = oneshot::channel(); let (tx, rx) = oneshot::channel();
task_spawner let initial_result = task_spawner
.spawn_async_with_rejection(Priority::P0, async move { .spawn_async_with_rejection_no_conversion(Priority::P0, async move {
let execution_layer = chain let execution_layer = chain
.execution_layer .execution_layer
.as_ref() .as_ref()
@ -3671,17 +3648,22 @@ pub fn serve<T: BeaconChainTypes>(
// from what is sent back down the channel. // from what is sent back down the channel.
Ok(warp::reply::reply().into_response()) Ok(warp::reply::reply().into_response())
}) })
.await?; .await;
if initial_result.is_err() {
return task_spawner::convert_rejection(initial_result).await;
}
// Await a response from the builder without blocking a // Await a response from the builder without blocking a
// `BeaconProcessor` worker. // `BeaconProcessor` worker.
rx.await.unwrap_or_else(|_| { task_spawner::convert_rejection(rx.await.unwrap_or_else(|_| {
Ok(warp::reply::with_status( Ok(warp::reply::with_status(
warp::reply::json(&"No response from channel"), warp::reply::json(&"No response from channel"),
eth2::StatusCode::INTERNAL_SERVER_ERROR, eth2::StatusCode::INTERNAL_SERVER_ERROR,
) )
.into_response()) .into_response())
}) }))
.await
}, },
); );
// POST validator/sync_committee_subscriptions // POST validator/sync_committee_subscriptions
@ -3694,7 +3676,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and(log_filter.clone()) .and(log_filter.clone())
.and_then( .then(
|subscriptions: Vec<types::SyncCommitteeSubscription>, |subscriptions: Vec<types::SyncCommitteeSubscription>,
validator_subscription_tx: Sender<ValidatorSubscriptionMessage>, validator_subscription_tx: Sender<ValidatorSubscriptionMessage>,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
@ -3738,7 +3720,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::body::json()) .and(warp::body::json())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and_then( .then(
|epoch: Epoch, |epoch: Epoch,
indices: Vec<u64>, indices: Vec<u64>,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
@ -3779,7 +3761,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::body::json()) .and(warp::body::json())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and_then( .then(
|request_data: api_types::LivenessRequestData, |request_data: api_types::LivenessRequestData,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>| { chain: Arc<BeaconChain<T>>| {
@ -3823,7 +3805,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path("health")) .and(warp::path("health"))
.and(warp::path::end()) .and(warp::path::end())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and_then(|task_spawner: TaskSpawner<T::EthSpec>| { .then(|task_spawner: TaskSpawner<T::EthSpec>| {
task_spawner.blocking_json_task(Priority::P0, move || { task_spawner.blocking_json_task(Priority::P0, move || {
eth2::lighthouse::Health::observe() eth2::lighthouse::Health::observe()
.map(api_types::GenericResponse::from) .map(api_types::GenericResponse::from)
@ -3841,7 +3823,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(app_start_filter) .and(app_start_filter)
.and(data_dir_filter) .and(data_dir_filter)
.and(network_globals.clone()) .and(network_globals.clone())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, |task_spawner: TaskSpawner<T::EthSpec>,
sysinfo, sysinfo,
app_start: std::time::Instant, app_start: std::time::Instant,
@ -3866,7 +3848,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::end()) .and(warp::path::end())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| { |task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| {
task_spawner.blocking_json_task(Priority::P1, move || { task_spawner.blocking_json_task(Priority::P1, move || {
ui::get_validator_count(chain).map(api_types::GenericResponse::from) ui::get_validator_count(chain).map(api_types::GenericResponse::from)
@ -3882,7 +3864,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::body::json()) .and(warp::body::json())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and_then( .then(
|request_data: ui::ValidatorMetricsRequestData, |request_data: ui::ValidatorMetricsRequestData,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>| { chain: Arc<BeaconChain<T>>| {
@ -3901,7 +3883,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::body::json()) .and(warp::body::json())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and_then( .then(
|request_data: ui::ValidatorInfoRequestData, |request_data: ui::ValidatorInfoRequestData,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>| { chain: Arc<BeaconChain<T>>| {
@ -3918,7 +3900,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::end()) .and(warp::path::end())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(network_globals.clone()) .and(network_globals.clone())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, |task_spawner: TaskSpawner<T::EthSpec>,
network_globals: Arc<NetworkGlobals<T::EthSpec>>| { network_globals: Arc<NetworkGlobals<T::EthSpec>>| {
task_spawner.blocking_json_task(Priority::P0, move || { task_spawner.blocking_json_task(Priority::P0, move || {
@ -3934,7 +3916,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path("nat")) .and(warp::path("nat"))
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(warp::path::end()) .and(warp::path::end())
.and_then(|task_spawner: TaskSpawner<T::EthSpec>| { .then(|task_spawner: TaskSpawner<T::EthSpec>| {
task_spawner.blocking_json_task(Priority::P1, move || { task_spawner.blocking_json_task(Priority::P1, move || {
Ok(api_types::GenericResponse::from( Ok(api_types::GenericResponse::from(
lighthouse_network::metrics::NAT_OPEN lighthouse_network::metrics::NAT_OPEN
@ -3952,7 +3934,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::end()) .and(warp::path::end())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(network_globals.clone()) .and(network_globals.clone())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, |task_spawner: TaskSpawner<T::EthSpec>,
network_globals: Arc<NetworkGlobals<T::EthSpec>>| { network_globals: Arc<NetworkGlobals<T::EthSpec>>| {
task_spawner.blocking_json_task(Priority::P1, move || { task_spawner.blocking_json_task(Priority::P1, move || {
@ -3976,7 +3958,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::end()) .and(warp::path::end())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(network_globals) .and(network_globals)
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, |task_spawner: TaskSpawner<T::EthSpec>,
network_globals: Arc<NetworkGlobals<T::EthSpec>>| { network_globals: Arc<NetworkGlobals<T::EthSpec>>| {
task_spawner.blocking_json_task(Priority::P1, move || { task_spawner.blocking_json_task(Priority::P1, move || {
@ -3999,7 +3981,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::end()) .and(warp::path::end())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| { |task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| {
task_spawner.blocking_response_task(Priority::P1, move || { task_spawner.blocking_response_task(Priority::P1, move || {
Ok::<_, warp::Rejection>(warp::reply::json( Ok::<_, warp::Rejection>(warp::reply::json(
@ -4023,7 +4005,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::end()) .and(warp::path::end())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and_then( .then(
|epoch: Epoch, |epoch: Epoch,
validator_id: ValidatorId, validator_id: ValidatorId,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
@ -4043,7 +4025,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::end()) .and(warp::path::end())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and_then( .then(
|epoch: Epoch, task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| { |epoch: Epoch, task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| {
task_spawner.blocking_json_task(Priority::P1, move || { task_spawner.blocking_json_task(Priority::P1, move || {
validator_inclusion::global_validator_inclusion_data(epoch, &chain) validator_inclusion::global_validator_inclusion_data(epoch, &chain)
@ -4059,7 +4041,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::end()) .and(warp::path::end())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| { |task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| {
task_spawner.blocking_json_task(Priority::P1, move || { task_spawner.blocking_json_task(Priority::P1, move || {
let current_slot_opt = chain.slot().ok(); let current_slot_opt = chain.slot().ok();
@ -4092,7 +4074,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::end()) .and(warp::path::end())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(eth1_service_filter.clone()) .and(eth1_service_filter.clone())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, eth1_service: eth1::Service| { |task_spawner: TaskSpawner<T::EthSpec>, eth1_service: eth1::Service| {
task_spawner.blocking_json_task(Priority::P1, move || { task_spawner.blocking_json_task(Priority::P1, move || {
Ok(api_types::GenericResponse::from( Ok(api_types::GenericResponse::from(
@ -4114,7 +4096,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::end()) .and(warp::path::end())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(eth1_service_filter) .and(eth1_service_filter)
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, eth1_service: eth1::Service| { |task_spawner: TaskSpawner<T::EthSpec>, eth1_service: eth1::Service| {
task_spawner.blocking_json_task(Priority::P1, move || { task_spawner.blocking_json_task(Priority::P1, move || {
Ok(api_types::GenericResponse::from( Ok(api_types::GenericResponse::from(
@ -4139,7 +4121,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::end()) .and(warp::path::end())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and_then( .then(
|state_id: StateId, |state_id: StateId,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>| { chain: Arc<BeaconChain<T>>| {
@ -4166,7 +4148,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::end()) .and(warp::path::end())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| { |task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| {
task_spawner.blocking_json_task(Priority::P1, move || { task_spawner.blocking_json_task(Priority::P1, move || {
if chain.eth1_chain.is_some() { if chain.eth1_chain.is_some() {
@ -4190,7 +4172,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::end()) .and(warp::path::end())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| { |task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| {
task_spawner.blocking_json_task(Priority::P1, move || database::info(chain)) task_spawner.blocking_json_task(Priority::P1, move || database::info(chain))
}, },
@ -4203,7 +4185,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(not_while_syncing_filter) .and(not_while_syncing_filter)
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| { |task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| {
task_spawner.blocking_json_task(Priority::P1, move || { task_spawner.blocking_json_task(Priority::P1, move || {
chain.store_migrator.process_reconstruction(); chain.store_migrator.process_reconstruction();
@ -4220,7 +4202,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and(log_filter.clone()) .and(log_filter.clone())
.and_then( .then(
|blocks: Vec<Arc<SignedBlindedBeaconBlock<T::EthSpec>>>, |blocks: Vec<Arc<SignedBlindedBeaconBlock<T::EthSpec>>>,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>, chain: Arc<BeaconChain<T>>,
@ -4246,7 +4228,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and(log_filter.clone()) .and(log_filter.clone())
.and_then(|query, task_spawner: TaskSpawner<T::EthSpec>, chain, log| { .then(|query, task_spawner: TaskSpawner<T::EthSpec>, chain, log| {
task_spawner.blocking_json_task(Priority::P1, move || { task_spawner.blocking_json_task(Priority::P1, move || {
block_rewards::get_block_rewards(query, chain, log) block_rewards::get_block_rewards(query, chain, log)
}) })
@ -4261,7 +4243,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and(log_filter.clone()) .and(log_filter.clone())
.and_then( .then(
|blocks, task_spawner: TaskSpawner<T::EthSpec>, chain, log| { |blocks, task_spawner: TaskSpawner<T::EthSpec>, chain, log| {
task_spawner.blocking_json_task(Priority::P1, move || { task_spawner.blocking_json_task(Priority::P1, move || {
block_rewards::compute_block_rewards(blocks, chain, log) block_rewards::compute_block_rewards(blocks, chain, log)
@ -4278,7 +4260,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::end()) .and(warp::path::end())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and_then( .then(
|target, query, task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| { |target, query, task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| {
task_spawner.blocking_json_task(Priority::P1, move || { task_spawner.blocking_json_task(Priority::P1, move || {
attestation_performance::get_attestation_performance(target, query, chain) attestation_performance::get_attestation_performance(target, query, chain)
@ -4294,7 +4276,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::end()) .and(warp::path::end())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and_then( .then(
|query, task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| { |query, task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| {
task_spawner.blocking_json_task(Priority::P1, move || { task_spawner.blocking_json_task(Priority::P1, move || {
block_packing_efficiency::get_block_packing_efficiency(query, chain) block_packing_efficiency::get_block_packing_efficiency(query, chain)
@ -4308,7 +4290,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::end()) .and(warp::path::end())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter.clone()) .and(chain_filter.clone())
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| { |task_spawner: TaskSpawner<T::EthSpec>, chain: Arc<BeaconChain<T>>| {
task_spawner.spawn_async_with_rejection(Priority::P1, async move { task_spawner.spawn_async_with_rejection(Priority::P1, async move {
let merge_readiness = chain.check_merge_readiness().await; let merge_readiness = chain.check_merge_readiness().await;
@ -4326,7 +4308,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(multi_key_query::<api_types::EventQuery>()) .and(multi_key_query::<api_types::EventQuery>())
.and(task_spawner_filter.clone()) .and(task_spawner_filter.clone())
.and(chain_filter) .and(chain_filter)
.and_then( .then(
|topics_res: Result<api_types::EventQuery, warp::Rejection>, |topics_res: Result<api_types::EventQuery, warp::Rejection>,
task_spawner: TaskSpawner<T::EthSpec>, task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>| { chain: Arc<BeaconChain<T>>| {
@ -4403,7 +4385,7 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::end()) .and(warp::path::end())
.and(task_spawner_filter) .and(task_spawner_filter)
.and(sse_component_filter) .and(sse_component_filter)
.and_then( .then(
|task_spawner: TaskSpawner<T::EthSpec>, sse_component: Option<SSELoggingComponents>| { |task_spawner: TaskSpawner<T::EthSpec>, sse_component: Option<SSELoggingComponents>| {
task_spawner.blocking_response_task(Priority::P1, move || { task_spawner.blocking_response_task(Priority::P1, move || {
if let Some(logging_components) = sse_component { if let Some(logging_components) = sse_component {

92
beacon_node/http_api/src/task_spawner.rs
View File

@ -35,6 +35,24 @@ pub struct TaskSpawner<E: EthSpec> {
beacon_processor_send: Option<BeaconProcessorSend<E>>, beacon_processor_send: Option<BeaconProcessorSend<E>>,
} }
/// Convert a warp `Rejection` into a `Response`.
///
/// This function should *always* be used to convert rejections into responses. This prevents warp
/// from trying to backtrack in strange ways. See: https://github.com/sigp/lighthouse/issues/3404
pub async fn convert_rejection<T: Reply>(res: Result<T, warp::Rejection>) -> Response {
match res {
Ok(response) => response.into_response(),
Err(e) => match warp_utils::reject::handle_rejection(e).await {
Ok(reply) => reply.into_response(),
Err(_) => warp::reply::with_status(
warp::reply::json(&"unhandled error"),
eth2::StatusCode::INTERNAL_SERVER_ERROR,
)
.into_response(),
},
}
}
impl<E: EthSpec> TaskSpawner<E> { impl<E: EthSpec> TaskSpawner<E> {
pub fn new(beacon_processor_send: Option<BeaconProcessorSend<E>>) -> Self { pub fn new(beacon_processor_send: Option<BeaconProcessorSend<E>>) -> Self {
Self { Self {
@ -43,11 +61,7 @@ impl<E: EthSpec> TaskSpawner<E> {
} }
/// Executes a "blocking" (non-async) task which returns a `Response`. /// Executes a "blocking" (non-async) task which returns a `Response`.
pub async fn blocking_response_task<F, T>( pub async fn blocking_response_task<F, T>(self, priority: Priority, func: F) -> Response
self,
priority: Priority,
func: F,
) -> Result<Response, warp::Rejection>
where where
F: FnOnce() -> Result<T, warp::Rejection> + Send + Sync + 'static, F: FnOnce() -> Result<T, warp::Rejection> + Send + Sync + 'static,
T: Reply + Send + 'static, T: Reply + Send + 'static,
@ -65,31 +79,25 @@ impl<E: EthSpec> TaskSpawner<E> {
}; };
// Send the function to the beacon processor for execution at some arbitrary time. // Send the function to the beacon processor for execution at some arbitrary time.
match send_to_beacon_processor( let result = send_to_beacon_processor(
beacon_processor_send, beacon_processor_send,
priority, priority,
BlockingOrAsync::Blocking(Box::new(process_fn)), BlockingOrAsync::Blocking(Box::new(process_fn)),
rx, rx,
) )
.await .await
{ .and_then(|x| x);
Ok(result) => result.map(Reply::into_response), convert_rejection(result).await
Err(error_response) => Ok(error_response),
}
} else { } else {
// There is no beacon processor so spawn a task directly on the // There is no beacon processor so spawn a task directly on the
// tokio executor. // tokio executor.
warp_utils::task::blocking_response_task(func).await convert_rejection(warp_utils::task::blocking_response_task(func).await).await
} }
} }
/// Executes a "blocking" (non-async) task which returns a JSON-serializable /// Executes a "blocking" (non-async) task which returns a JSON-serializable
/// object. /// object.
pub async fn blocking_json_task<F, T>( pub async fn blocking_json_task<F, T>(self, priority: Priority, func: F) -> Response
self,
priority: Priority,
func: F,
) -> Result<Response, warp::Rejection>
where where
F: FnOnce() -> Result<T, warp::Rejection> + Send + Sync + 'static, F: FnOnce() -> Result<T, warp::Rejection> + Send + Sync + 'static,
T: Serialize + Send + 'static, T: Serialize + Send + 'static,
@ -98,11 +106,26 @@ impl<E: EthSpec> TaskSpawner<E> {
self.blocking_response_task(priority, func).await self.blocking_response_task(priority, func).await
} }
/// Executes an async task which may return a `warp::Rejection`. /// Executes an async task which may return a `Rejection`, which will be converted to a response.
pub async fn spawn_async_with_rejection( pub async fn spawn_async_with_rejection(
self, self,
priority: Priority, priority: Priority,
func: impl Future<Output = Result<Response, warp::Rejection>> + Send + Sync + 'static, func: impl Future<Output = Result<Response, warp::Rejection>> + Send + Sync + 'static,
) -> Response {
let result = self
.spawn_async_with_rejection_no_conversion(priority, func)
.await;
convert_rejection(result).await
}
/// Same as `spawn_async_with_rejection` but returning a result with the unhandled rejection.
///
/// If you call this function you MUST convert the rejection to a response and not let it
/// propagate into Warp's filters. See `convert_rejection`.
pub async fn spawn_async_with_rejection_no_conversion(
self,
priority: Priority,
func: impl Future<Output = Result<Response, warp::Rejection>> + Send + Sync + 'static,
) -> Result<Response, warp::Rejection> { ) -> Result<Response, warp::Rejection> {
if let Some(beacon_processor_send) = &self.beacon_processor_send { if let Some(beacon_processor_send) = &self.beacon_processor_send {
// Create a wrapper future that will execute `func` and send the // Create a wrapper future that will execute `func` and send the
@ -124,18 +147,16 @@ impl<E: EthSpec> TaskSpawner<E> {
rx, rx,
) )
.await .await
.unwrap_or_else(Result::Ok) .and_then(|x| x)
} else { } else {
// There is no beacon processor so spawn a task directly on the // There is no beacon processor so spawn a task directly on the
// tokio executor. // tokio executor.
tokio::task::spawn(func).await.unwrap_or_else(|e| { tokio::task::spawn(func)
let response = warp::reply::with_status( .await
warp::reply::json(&format!("Tokio did not execute task: {e:?}")), .map_err(|_| {
eth2::StatusCode::INTERNAL_SERVER_ERROR, warp_utils::reject::custom_server_error("Tokio failed to spawn task".into())
) })
.into_response(); .and_then(|x| x)
Ok(response)
})
} }
} }
@ -158,14 +179,14 @@ impl<E: EthSpec> TaskSpawner<E> {
}; };
// Send the function to the beacon processor for execution at some arbitrary time. // Send the function to the beacon processor for execution at some arbitrary time.
send_to_beacon_processor( let result = send_to_beacon_processor(
beacon_processor_send, beacon_processor_send,
priority, priority,
BlockingOrAsync::Async(Box::pin(process_fn)), BlockingOrAsync::Async(Box::pin(process_fn)),
rx, rx,
) )
.await .await;
.unwrap_or_else(|error_response| error_response) convert_rejection(result).await
} else { } else {
// There is no beacon processor so spawn a task directly on the // There is no beacon processor so spawn a task directly on the
// tokio executor. // tokio executor.
@ -182,14 +203,14 @@ impl<E: EthSpec> TaskSpawner<E> {
/// Send a task to the beacon processor and await execution. /// Send a task to the beacon processor and await execution.
/// ///
/// If the task is not executed, return an `Err(response)` with an error message /// If the task is not executed, return an `Err` with an error message
/// for the API consumer. /// for the API consumer.
async fn send_to_beacon_processor<E: EthSpec, T>( async fn send_to_beacon_processor<E: EthSpec, T>(
beacon_processor_send: &BeaconProcessorSend<E>, beacon_processor_send: &BeaconProcessorSend<E>,
priority: Priority, priority: Priority,
process_fn: BlockingOrAsync, process_fn: BlockingOrAsync,
rx: oneshot::Receiver<T>, rx: oneshot::Receiver<T>,
) -> Result<T, Response> { ) -> Result<T, warp::Rejection> {
let error_message = match beacon_processor_send.try_send(priority.work_event(process_fn)) { let error_message = match beacon_processor_send.try_send(priority.work_event(process_fn)) {
Ok(()) => { Ok(()) => {
match rx.await { match rx.await {
@ -205,10 +226,7 @@ async fn send_to_beacon_processor<E: EthSpec, T>(
Err(TrySendError::Closed(_)) => "The task was dropped. The server is shutting down.", Err(TrySendError::Closed(_)) => "The task was dropped. The server is shutting down.",
}; };
let error_response = warp::reply::with_status( Err(warp_utils::reject::custom_server_error(
warp::reply::json(&error_message), error_message.to_string(),
eth2::StatusCode::INTERNAL_SERVER_ERROR, ))
)
.into_response();
Err(error_response)
} }