Skip to content

Commit

Permalink
Fix http_api, make lint passes
Browse files Browse the repository at this point in the history
  • Loading branch information
@paulhauner
paulhauner committed Jun 28, 2022
1 parent 0a6b9ae commit 3e21adb
Showing 1 changed file with 149 additions and 165 deletions.
314 changes: 149 additions & 165 deletions beacon_node/http_api/src/lib.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -911,96 +911,85 @@ pub fn serve<T: BeaconChainTypes>(
.and(network_tx_filter.clone())
.and(log_filter.clone())
.and_then(
|block: SignedBeaconBlock<T::EthSpec>,
|block: Arc<SignedBeaconBlock<T::EthSpec>>,
chain: Arc<BeaconChain<T>>,
network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>,
log: Logger| {
blocking_json_task(move || {
let seen_timestamp = timestamp_now();

// Send the block, regardless of whether or not it is valid. The API
// specification is very clear that this is the desired behaviour.
publish_pubsub_message(
&network_tx,
PubsubMessage::BeaconBlock(Box::new(block.clone())),
)?;
log: Logger| async move {
let seen_timestamp = timestamp_now();

// Send the block, regardless of whether or not it is valid. The API
// specification is very clear that this is the desired behaviour.
publish_pubsub_message(&network_tx, PubsubMessage::BeaconBlock(block.clone()))?;

// Determine the delay after the start of the slot, register it with metrics.
let delay = get_block_delay_ms(seen_timestamp, block.message(), &chain.slot_clock);
metrics::observe_duration(&metrics::HTTP_API_BLOCK_BROADCAST_DELAY_TIMES, delay);

match chain.process_block(block.clone()).await {
Ok(root) => {
info!(
log,
"Valid block from HTTP API";
"block_delay" => ?delay,
"root" => format!("{}", root),
"proposer_index" => block.message().proposer_index(),
"slot" => block.slot(),
);

// Determine the delay after the start of the slot, register it with metrics.
let delay =
get_block_delay_ms(seen_timestamp, block.message(), &chain.slot_clock);
metrics::observe_duration(
&metrics::HTTP_API_BLOCK_BROADCAST_DELAY_TIMES,
delay,
);
// Notify the validator monitor.
chain.validator_monitor.read().register_api_block(
seen_timestamp,
block.message(),
root,
&chain.slot_clock,
);

match chain.process_block(block.clone()) {
Ok(root) => {
info!(
// Update the head since it's likely this block will become the new
// head.
chain
.recompute_head_at_current_slot()
.await
.map_err(warp_utils::reject::beacon_chain_error)?;

// Perform some logging to inform users if their blocks are being produced
// late.
//
// Check to see the thresholds are non-zero to avoid logging errors with small
// slot times (e.g., during testing)
let crit_threshold = chain.slot_clock.unagg_attestation_production_delay();
let error_threshold = crit_threshold / 2;
if delay >= crit_threshold {
crit!(
log,
"Valid block from HTTP API";
"block_delay" => ?delay,
"root" => format!("{}", root),
"proposer_index" => block.message().proposer_index(),
"Block was broadcast too late";
"msg" => "system may be overloaded, block likely to be orphaned",
"delay_ms" => delay.as_millis(),
"slot" => block.slot(),
);

// Notify the validator monitor.
chain.validator_monitor.read().register_api_block(
seen_timestamp,
block.message(),
root,
&chain.slot_clock,
);

// Update the head since it's likely this block will become the new
// head.
//
// This is spawned in a new task since it's not imperative we wait for
// it to complete.
chain
.clone()
.spawn_recompute_head_at_current_slot("api_full_block");

// Perform some logging to inform users if their blocks are being produced
// late.
//
// Check to see the thresholds are non-zero to avoid logging errors with small
// slot times (e.g., during testing)
let crit_threshold = chain.slot_clock.unagg_attestation_production_delay();
let error_threshold = crit_threshold / 2;
if delay >= crit_threshold {
crit!(
log,
"Block was broadcast too late";
"msg" => "system may be overloaded, block likely to be orphaned",
"delay_ms" => delay.as_millis(),
"slot" => block.slot(),
"root" => ?root,
)
} else if delay >= error_threshold {
error!(
log,
"Block broadcast was delayed";
"msg" => "system may be overloaded, block may be orphaned",
"delay_ms" => delay.as_millis(),
"slot" => block.slot(),
"root" => ?root,
)
}

Ok(())
}
Err(e) => {
let msg = format!("{:?}", e);
"root" => ?root,
)
} else if delay >= error_threshold {
error!(
log,
"Invalid block provided to HTTP API";
"reason" => &msg
);
Err(warp_utils::reject::broadcast_without_import(msg))
"Block broadcast was delayed";
"msg" => "system may be overloaded, block may be orphaned",
"delay_ms" => delay.as_millis(),
"slot" => block.slot(),
"root" => ?root,
)
}

Ok(warp::reply::json(&()))
}
})
Err(e) => {
let msg = format!("{:?}", e);
error!(
log,
"Invalid block provided to HTTP API";
"reason" => &msg
);
Err(warp_utils::reject::broadcast_without_import(msg))
}
}
},
);

Expand All @@ -1018,100 +1007,95 @@ pub fn serve<T: BeaconChainTypes>(
.and(network_tx_filter.clone())
.and(log_filter.clone())
.and_then(
|block: SignedBeaconBlock<T::EthSpec, BlindedPayload<_>>,
|block: Arc<SignedBeaconBlock<T::EthSpec, BlindedPayload<_>>>,
chain: Arc<BeaconChain<T>>,
network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>,
_log: Logger| {
blocking_json_task(move || {
if let Some(el) = chain.execution_layer.as_ref() {
//FIXME(sean): we may not always receive the payload in this response because it
// should be the relay's job to propogate the block. However, since this block is
// already signed and sent this might be ok (so long as the relay validates
// the block before revealing the payload).

//FIXME(sean) additionally, this endpoint should serve blocks prior to Bellatrix, and should
// be able to support the normal block proposal flow, because at some point full block endpoints
// will be deprecated from the beacon API. This will entail creating full blocks in
// `validator/blinded_blocks`, caching their payloads, and transforming them into blinded
// blocks. We will access the payload of those blocks here. This flow should happen if the
// execution layer has no payload builders or if we have not yet finalized post-merge transition.
let payload = el
.block_on(|el| el.propose_blinded_beacon_block(&block))
.map_err(|e| {
warp_utils::reject::custom_server_error(format!(
"proposal failed: {:?}",
e
))
})?;
let new_block = SignedBeaconBlock::Merge(SignedBeaconBlockMerge {
message: BeaconBlockMerge {
slot: block.message().slot(),
proposer_index: block.message().proposer_index(),
parent_root: block.message().parent_root(),
state_root: block.message().state_root(),
body: BeaconBlockBodyMerge {
randao_reveal: block.message().body().randao_reveal().clone(),
eth1_data: block.message().body().eth1_data().clone(),
graffiti: *block.message().body().graffiti(),
proposer_slashings: block
.message()
.body()
.proposer_slashings()
.clone(),
attester_slashings: block
.message()
.body()
.attester_slashings()
.clone(),
attestations: block.message().body().attestations().clone(),
deposits: block.message().body().deposits().clone(),
voluntary_exits: block
.message()
.body()
.voluntary_exits()
.clone(),
sync_aggregate: block
.message()
.body()
.sync_aggregate()
.unwrap()
.clone(),
execution_payload: payload.into(),
},
_log: Logger| async move {
if let Some(el) = chain.execution_layer.as_ref() {
//FIXME(sean): we may not always receive the payload in this response because it
// should be the relay's job to propogate the block. However, since this block is
// already signed and sent this might be ok (so long as the relay validates
// the block before revealing the payload).

//FIXME(sean) additionally, this endpoint should serve blocks prior to Bellatrix, and should
// be able to support the normal block proposal flow, because at some point full block endpoints
// will be deprecated from the beacon API. This will entail creating full blocks in
// `validator/blinded_blocks`, caching their payloads, and transforming them into blinded
// blocks. We will access the payload of those blocks here. This flow should happen if the
// execution layer has no payload builders or if we have not yet finalized post-merge transition.
let payload = el
.block_on(|el| el.propose_blinded_beacon_block(&block))
.map_err(|e| {
warp_utils::reject::custom_server_error(format!(
"proposal failed: {:?}",
e
))
})?;
let new_block = SignedBeaconBlock::Merge(SignedBeaconBlockMerge {
message: BeaconBlockMerge {
slot: block.message().slot(),
proposer_index: block.message().proposer_index(),
parent_root: block.message().parent_root(),
state_root: block.message().state_root(),
body: BeaconBlockBodyMerge {
randao_reveal: block.message().body().randao_reveal().clone(),
eth1_data: block.message().body().eth1_data().clone(),
graffiti: *block.message().body().graffiti(),
proposer_slashings: block
.message()
.body()
.proposer_slashings()
.clone(),
attester_slashings: block
.message()
.body()
.attester_slashings()
.clone(),
attestations: block.message().body().attestations().clone(),
deposits: block.message().body().deposits().clone(),
voluntary_exits: block.message().body().voluntary_exits().clone(),
sync_aggregate: block
.message()
.body()
.sync_aggregate()
.unwrap()
.clone(),
execution_payload: payload.into(),
},
signature: block.signature().clone(),
});
},
signature: block.signature().clone(),
});
let new_block = Arc::new(new_block);

// Send the block, regardless of whether or not it is valid. The API
// specification is very clear that this is the desired behaviour.
publish_pubsub_message(
&network_tx,
PubsubMessage::BeaconBlock(Box::new(new_block.clone())),
)?;
// Send the block, regardless of whether or not it is valid. The API
// specification is very clear that this is the desired behaviour.
publish_pubsub_message(
&network_tx,
PubsubMessage::BeaconBlock(new_block.clone()),
)?;

match chain.process_block(new_block) {
Ok(_) => {
// Update the head since it's likely this block will become the new
// head.
//
// This is spawned in a new task since it's not imperative we wait for
// it to complete.
chain.spawn_recompute_head_at_current_slot("api_blinded_block");
match chain.process_block(new_block).await {
Ok(_) => {
// Update the head since it's likely this block will become the new
// head.
chain
.recompute_head_at_current_slot()
.await
.map_err(warp_utils::reject::beacon_chain_error)?;

Ok(())
}
Err(e) => {
let msg = format!("{:?}", e);
Ok(warp::reply::json(&()))
}
Err(e) => {
let msg = format!("{:?}", e);

Err(warp_utils::reject::broadcast_without_import(msg))
}
Err(warp_utils::reject::broadcast_without_import(msg))
}
} else {
Err(warp_utils::reject::custom_server_error(
"no execution layer found".to_string(),
))
}
})
} else {
Err(warp_utils::reject::custom_server_error(
"no execution layer found".to_string(),
))
}
},
);

Expand Down

0 comments on commit 3e21adb

Please sign in to comment.