Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Handle unknown head during attestation publishing #5010

Merged
merged 8 commits into from
Feb 15, 2024
Merged

Handle unknown head during attestation publishing #5010

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
ab70bbf
Handle unknown head during attestation publishing
michaelsproul Dec 13, 2023
f33d8c2
Merge remote-tracking branch 'origin/unstable' into queue-http-attest…
michaelsproul Feb 7, 2024
952e87c
Simplify task spawner
michaelsproul Feb 7, 2024
49862f8
Improve logging
michaelsproul Feb 7, 2024
c505174
Add a test
michaelsproul Feb 9, 2024
6043b0c
Improve error logging
michaelsproul Feb 14, 2024
cf60080
Merge remote-tracking branch 'origin/unstable' into queue-http-attest…
michaelsproul Feb 14, 2024
26a62fb
Fix beta compiler warnings
michaelsproul Feb 14, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions beacon_node/client/src/builder.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -505,6 +505,7 @@ where
network_senders: None,
network_globals: None,
beacon_processor_send: None,
beacon_processor_reprocess_send: None,
eth1_service: Some(genesis_service.eth1_service.clone()),
log: context.log().clone(),
sse_logging_components: runtime_context.sse_logging_components.clone(),
Expand Down Expand Up @@ -747,6 +748,9 @@ where
network_globals: self.network_globals.clone(),
eth1_service: self.eth1_service.clone(),
beacon_processor_send: Some(beacon_processor_channels.beacon_processor_tx.clone()),
beacon_processor_reprocess_send: Some(
beacon_processor_channels.work_reprocessing_tx.clone(),
),
sse_logging_components: runtime_context.sse_logging_components.clone(),
log: log.clone(),
});
Expand Down
151 changes: 22 additions & 129 deletions beacon_node/http_api/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ mod database;
mod metrics;
mod produce_block;
mod proposer_duties;
mod publish_attestations;
mod publish_blocks;
mod standard_block_rewards;
mod state_id;
Expand All @@ -35,7 +36,7 @@ use beacon_chain::{
validator_monitor::timestamp_now, AttestationError as AttnError, BeaconChain, BeaconChainError,
BeaconChainTypes, WhenSlotSkipped,
};
use beacon_processor::BeaconProcessorSend;
use beacon_processor::{work_reprocessing_queue::ReprocessQueueMessage, BeaconProcessorSend};
pub use block_id::BlockId;
use builder_states::get_next_withdrawals;
use bytes::Bytes;
Expand Down Expand Up @@ -129,6 +130,7 @@ pub struct Context<T: BeaconChainTypes> {
pub network_senders: Option<NetworkSenders<T::EthSpec>>,
pub network_globals: Option<Arc<NetworkGlobals<T::EthSpec>>>,
pub beacon_processor_send: Option<BeaconProcessorSend<T::EthSpec>>,
pub beacon_processor_reprocess_send: Option<Sender<ReprocessQueueMessage>>,
pub eth1_service: Option<eth1::Service>,
pub sse_logging_components: Option<SSELoggingComponents>,
pub log: Logger,
Expand Down Expand Up @@ -534,6 +536,11 @@ pub fn serve<T: BeaconChainTypes>(
.filter(|_| config.enable_beacon_processor);
let task_spawner_filter =
warp::any().map(move || TaskSpawner::new(beacon_processor_send.clone()));
let beacon_processor_reprocess_send = ctx
.beacon_processor_reprocess_send
.clone()
.filter(|_| config.enable_beacon_processor);
let reprocess_send_filter = warp::any().map(move || beacon_processor_reprocess_send.clone());

let duplicate_block_status_code = ctx.config.duplicate_block_status_code;

Expand Down Expand Up @@ -1756,140 +1763,26 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::path::end())
.and(warp_utils::json::json())
.and(network_tx_filter.clone())
.and(reprocess_send_filter)
.and(log_filter.clone())
.then(
|task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>,
attestations: Vec<Attestation<T::EthSpec>>,
network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>,
log: Logger| {
task_spawner.blocking_json_task(Priority::P0, move || {
let seen_timestamp = timestamp_now();
let mut failures = Vec::new();
let mut num_already_known = 0;

for (index, attestation) in attestations.as_slice().iter().enumerate() {
let attestation = match chain
.verify_unaggregated_attestation_for_gossip(attestation, None)
{
Ok(attestation) => attestation,
Err(AttnError::PriorAttestationKnown { .. }) => {
num_already_known += 1;

// Skip to the next attestation since an attestation for this
// validator is already known in this epoch.
//
// There's little value for the network in validating a second
// attestation for another validator since it is either:
//
// 1. A duplicate.
// 2. Slashable.
// 3. Invalid.
//
// We are likely to get duplicates in the case where a VC is using
// fallback BNs. If the first BN actually publishes some/all of a
// batch of attestations but fails to respond in a timely fashion,
// the VC is likely to try publishing the attestations on another
// BN. That second BN may have already seen the attestations from
// the first BN and therefore indicate that the attestations are
// "already seen". An attestation that has already been seen has
// been published on the network so there's no actual error from
// the perspective of the user.
//
// It's better to prevent slashable attestations from ever
// appearing on the network than trying to slash validators,
// especially those validators connected to the local API.
//
// There might be *some* value in determining that this attestation
// is invalid, but since a valid attestation already it exists it
// appears that this validator is capable of producing valid
// attestations and there's no immediate cause for concern.
continue;
}
Err(e) => {
error!(log,
"Failure verifying attestation for gossip";
"error" => ?e,
"request_index" => index,
"committee_index" => attestation.data.index,
"attestation_slot" => attestation.data.slot,
);
failures.push(api_types::Failure::new(
index,
format!("Verification: {:?}", e),
));
// skip to the next attestation so we do not publish this one to gossip
continue;
}
};

// Notify the validator monitor.
chain
.validator_monitor
.read()
.register_api_unaggregated_attestation(
seen_timestamp,
attestation.indexed_attestation(),
&chain.slot_clock,
);

publish_pubsub_message(
&network_tx,
PubsubMessage::Attestation(Box::new((
attestation.subnet_id(),
attestation.attestation().clone(),
))),
)?;

let committee_index = attestation.attestation().data.index;
let slot = attestation.attestation().data.slot;

if let Err(e) = chain.apply_attestation_to_fork_choice(&attestation) {
error!(log,
"Failure applying verified attestation to fork choice";
"error" => ?e,
"request_index" => index,
"committee_index" => committee_index,
"slot" => slot,
);
failures.push(api_types::Failure::new(
index,
format!("Fork choice: {:?}", e),
));
};

if let Err(e) = chain.add_to_naive_aggregation_pool(&attestation) {
error!(log,
"Failure adding verified attestation to the naive aggregation pool";
"error" => ?e,
"request_index" => index,
"committee_index" => committee_index,
"slot" => slot,
);
failures.push(api_types::Failure::new(
index,
format!("Naive aggregation pool: {:?}", e),
));
}
}

if num_already_known > 0 {
debug!(
log,
"Some unagg attestations already known";
"count" => num_already_known
);
}

if failures.is_empty() {
Ok(())
} else {
Err(warp_utils::reject::indexed_bad_request(
"error processing attestations".to_string(),
failures,
))
}
})
reprocess_tx: Option<Sender<ReprocessQueueMessage>>,
log: Logger| async move {
let result = crate::publish_attestations::publish_attestations(
task_spawner,
chain,
attestations,
network_tx,
reprocess_tx,
log,
)
.await
.map(|()| warp::reply::json(&()));
task_spawner::convert_rejection(result).await
},
);

Expand Down
Loading
Loading