mirror of
https://github.com/sigp/lighthouse.git
synced 2026-07-03 12:54:27 +00:00
@@ -1188,22 +1188,14 @@ impl<T: BeaconChainTypes> SyncManager<T> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn on_sampling_result(&mut self, requester: SamplingRequester, result: SamplingResult) {
|
fn on_sampling_result(&mut self, requester: SamplingRequester, result: SamplingResult) {
|
||||||
// TODO(das): How is a consumer of sampling results?
|
|
||||||
// - Fork-choice for trailing DA
|
|
||||||
// - Single lookups to complete import requirements
|
|
||||||
// - Range sync to complete import requirements? Can sampling for syncing lag behind and
|
|
||||||
// accumulate in fork-choice?
|
|
||||||
|
|
||||||
match requester {
|
match requester {
|
||||||
SamplingRequester::ImportedBlock(block_root) => {
|
SamplingRequester::ImportedBlock(block_root) => {
|
||||||
debug!(self.log, "Sampling result"; "block_root" => %block_root, "result" => ?result);
|
debug!(self.log, "Sampling result"; "block_root" => %block_root, "result" => ?result);
|
||||||
|
|
||||||
// TODO(das): Consider moving SamplingResult to the beacon_chain crate and import
|
|
||||||
// here. No need to add too much enum variants, just whatever the beacon_chain or
|
|
||||||
// fork-choice needs to make a decision. Currently the fork-choice only needs to
|
|
||||||
// be notified of successful samplings, i.e. sampling failures don't trigger pruning
|
|
||||||
match result {
|
match result {
|
||||||
Ok(_) => {
|
Ok(_) => {
|
||||||
|
// Notify the fork-choice of a successful sampling result to mark the block
|
||||||
|
// branch as safe.
|
||||||
if let Err(e) = self
|
if let Err(e) = self
|
||||||
.network
|
.network
|
||||||
.beacon_processor()
|
.beacon_processor()
|
||||||
|
|||||||
@@ -769,7 +769,6 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
|
|||||||
self.log.clone(),
|
self.log.clone(),
|
||||||
);
|
);
|
||||||
|
|
||||||
// TODO(das): start request
|
|
||||||
// Note that you can only send, but not handle a response here
|
// Note that you can only send, but not handle a response here
|
||||||
match request.continue_requests(self) {
|
match request.continue_requests(self) {
|
||||||
Ok(_) => {
|
Ok(_) => {
|
||||||
@@ -779,7 +778,6 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
|
|||||||
self.custody_by_root_requests.insert(requester, request);
|
self.custody_by_root_requests.insert(requester, request);
|
||||||
Ok(LookupRequestResult::RequestSent(req_id))
|
Ok(LookupRequestResult::RequestSent(req_id))
|
||||||
}
|
}
|
||||||
// TODO(das): handle this error properly
|
|
||||||
Err(e) => Err(RpcRequestSendError::CustodyRequestError(e)),
|
Err(e) => Err(RpcRequestSendError::CustodyRequestError(e)),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -24,7 +24,6 @@ pub type SamplingResult = Result<(), SamplingError>;
|
|||||||
type DataColumnSidecarList<E> = Vec<Arc<DataColumnSidecar<E>>>;
|
type DataColumnSidecarList<E> = Vec<Arc<DataColumnSidecar<E>>>;
|
||||||
|
|
||||||
pub struct Sampling<T: BeaconChainTypes> {
|
pub struct Sampling<T: BeaconChainTypes> {
|
||||||
// TODO(das): stalled sampling request are never cleaned up
|
|
||||||
requests: HashMap<SamplingRequester, ActiveSamplingRequest<T>>,
|
requests: HashMap<SamplingRequester, ActiveSamplingRequest<T>>,
|
||||||
sampling_config: SamplingConfig,
|
sampling_config: SamplingConfig,
|
||||||
log: slog::Logger,
|
log: slog::Logger,
|
||||||
@@ -313,8 +312,8 @@ impl<T: BeaconChainTypes> ActiveSamplingRequest<T> {
|
|||||||
.iter()
|
.iter()
|
||||||
.position(|data| &data.index == column_index)
|
.position(|data| &data.index == column_index)
|
||||||
else {
|
else {
|
||||||
// Peer does not have the requested data.
|
// Peer does not have the requested data, mark peer as "dont have" and try
|
||||||
// TODO(das) what to do?
|
// again with a different peer.
|
||||||
debug!(self.log,
|
debug!(self.log,
|
||||||
"Sampling peer claims to not have the data";
|
"Sampling peer claims to not have the data";
|
||||||
"block_root" => %self.block_root,
|
"block_root" => %self.block_root,
|
||||||
@@ -373,7 +372,9 @@ impl<T: BeaconChainTypes> ActiveSamplingRequest<T> {
|
|||||||
sampling_request_id,
|
sampling_request_id,
|
||||||
},
|
},
|
||||||
) {
|
) {
|
||||||
// TODO(das): Beacon processor is overloaded, what should we do?
|
// Beacon processor is overloaded, drop sampling attempt. Failing to sample
|
||||||
|
// is not a permanent state so we should recover once the node has capacity
|
||||||
|
// and receives a descendant block.
|
||||||
error!(self.log,
|
error!(self.log,
|
||||||
"Dropping sampling";
|
"Dropping sampling";
|
||||||
"block" => %self.block_root,
|
"block" => %self.block_root,
|
||||||
@@ -391,8 +392,8 @@ impl<T: BeaconChainTypes> ActiveSamplingRequest<T> {
|
|||||||
);
|
);
|
||||||
metrics::inc_counter_vec(&metrics::SAMPLE_DOWNLOAD_RESULT, &[metrics::FAILURE]);
|
metrics::inc_counter_vec(&metrics::SAMPLE_DOWNLOAD_RESULT, &[metrics::FAILURE]);
|
||||||
|
|
||||||
// Error downloading, maybe penalize peer and retry again.
|
// Error downloading, malicious network errors are already penalized before
|
||||||
// TODO(das) with different peer or different peer?
|
// reaching this function. Mark the peer as failed and try again with another.
|
||||||
for column_index in column_indexes {
|
for column_index in column_indexes {
|
||||||
let Some(request) = self.column_requests.get_mut(column_index) else {
|
let Some(request) = self.column_requests.get_mut(column_index) else {
|
||||||
warn!(self.log,
|
warn!(self.log,
|
||||||
@@ -453,7 +454,7 @@ impl<T: BeaconChainTypes> ActiveSamplingRequest<T> {
|
|||||||
debug!(self.log, "Sample verification failure"; "block_root" => %self.block_root, "column_indexes" => ?column_indexes, "reason" => ?err);
|
debug!(self.log, "Sample verification failure"; "block_root" => %self.block_root, "column_indexes" => ?column_indexes, "reason" => ?err);
|
||||||
metrics::inc_counter_vec(&metrics::SAMPLE_VERIFY_RESULT, &[metrics::FAILURE]);
|
metrics::inc_counter_vec(&metrics::SAMPLE_VERIFY_RESULT, &[metrics::FAILURE]);
|
||||||
|
|
||||||
// TODO(das): Peer sent invalid data, penalize and try again from different peer
|
// Peer sent invalid data, penalize and try again from different peer
|
||||||
// TODO(das): Count individual failures
|
// TODO(das): Count individual failures
|
||||||
for column_index in column_indexes {
|
for column_index in column_indexes {
|
||||||
let Some(request) = self.column_requests.get_mut(column_index) else {
|
let Some(request) = self.column_requests.get_mut(column_index) else {
|
||||||
|
|||||||
Reference in New Issue
Block a user