mirror of
https://github.com/sigp/lighthouse.git
synced 2026-05-07 16:55:46 +00:00
Activate clippy::manual_let_else lint (#4889)
## Issue Addressed #4888 ## Proposed Changes Enabled `clippy::manual_let_else` lint and resolved the warning messages.
This commit is contained in:
@@ -601,11 +601,11 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
|
||||
spec: &ChainSpec,
|
||||
log: &Logger,
|
||||
) -> Result<Option<BeaconForkChoice<T>>, Error> {
|
||||
let persisted_fork_choice =
|
||||
match store.get_item::<PersistedForkChoice>(&FORK_CHOICE_DB_KEY)? {
|
||||
Some(fc) => fc,
|
||||
None => return Ok(None),
|
||||
};
|
||||
let Some(persisted_fork_choice) =
|
||||
store.get_item::<PersistedForkChoice>(&FORK_CHOICE_DB_KEY)?
|
||||
else {
|
||||
return Ok(None);
|
||||
};
|
||||
|
||||
let fc_store =
|
||||
BeaconForkChoiceStore::from_persisted(persisted_fork_choice.fork_choice_store, store)?;
|
||||
@@ -3485,9 +3485,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
|
||||
state: &BeaconState<T::EthSpec>,
|
||||
) -> Result<(), BlockError<T::EthSpec>> {
|
||||
// Only perform the weak subjectivity check if it was configured.
|
||||
let wss_checkpoint = if let Some(checkpoint) = self.config.weak_subjectivity_checkpoint {
|
||||
checkpoint
|
||||
} else {
|
||||
let Some(wss_checkpoint) = self.config.weak_subjectivity_checkpoint else {
|
||||
return Ok(());
|
||||
};
|
||||
// Note: we're using the finalized checkpoint from the head state, rather than fork
|
||||
@@ -5336,14 +5334,11 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
|
||||
)
|
||||
.await??;
|
||||
|
||||
let (forkchoice_update_params, pre_payload_attributes) =
|
||||
if let Some((fcu, Some(pre_payload))) = maybe_prep_data {
|
||||
(fcu, pre_payload)
|
||||
} else {
|
||||
// Appropriate log messages have already been logged above and in
|
||||
// `get_pre_payload_attributes`.
|
||||
return Ok(());
|
||||
};
|
||||
let Some((forkchoice_update_params, Some(pre_payload_attributes))) = maybe_prep_data else {
|
||||
// Appropriate log messages have already been logged above and in
|
||||
// `get_pre_payload_attributes`.
|
||||
return Ok(());
|
||||
};
|
||||
|
||||
// If the execution layer doesn't have any proposer data for this validator then we assume
|
||||
// it's not connected to this BN and no action is required.
|
||||
@@ -5436,23 +5431,20 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
|
||||
}
|
||||
}
|
||||
|
||||
let till_prepare_slot =
|
||||
if let Some(duration) = self.slot_clock.duration_to_slot(prepare_slot) {
|
||||
duration
|
||||
} else {
|
||||
// `SlotClock::duration_to_slot` will return `None` when we are past the start
|
||||
// of `prepare_slot`. Don't bother sending a `forkchoiceUpdated` in that case,
|
||||
// it's too late.
|
||||
//
|
||||
// This scenario might occur on an overloaded/under-resourced node.
|
||||
warn!(
|
||||
self.log,
|
||||
"Delayed proposer preparation";
|
||||
"prepare_slot" => prepare_slot,
|
||||
"validator" => proposer,
|
||||
);
|
||||
return Ok(());
|
||||
};
|
||||
let Some(till_prepare_slot) = self.slot_clock.duration_to_slot(prepare_slot) else {
|
||||
// `SlotClock::duration_to_slot` will return `None` when we are past the start
|
||||
// of `prepare_slot`. Don't bother sending a `forkchoiceUpdated` in that case,
|
||||
// it's too late.
|
||||
//
|
||||
// This scenario might occur on an overloaded/under-resourced node.
|
||||
warn!(
|
||||
self.log,
|
||||
"Delayed proposer preparation";
|
||||
"prepare_slot" => prepare_slot,
|
||||
"validator" => proposer,
|
||||
);
|
||||
return Ok(());
|
||||
};
|
||||
|
||||
// If we are close enough to the proposal slot, send an fcU, which will have payload
|
||||
// attributes filled in by the execution layer cache we just primed.
|
||||
|
||||
@@ -451,23 +451,21 @@ async fn availability_cache_maintenance_service<T: BeaconChainTypes>(
|
||||
let additional_delay = (epoch_duration * 3) / 4;
|
||||
tokio::time::sleep(duration + additional_delay).await;
|
||||
|
||||
let deneb_fork_epoch = match chain.spec.deneb_fork_epoch {
|
||||
Some(epoch) => epoch,
|
||||
None => break, // shutdown service if deneb fork epoch not set
|
||||
let Some(deneb_fork_epoch) = chain.spec.deneb_fork_epoch else {
|
||||
// shutdown service if deneb fork epoch not set
|
||||
break;
|
||||
};
|
||||
|
||||
debug!(
|
||||
chain.log,
|
||||
"Availability cache maintenance service firing";
|
||||
);
|
||||
|
||||
let current_epoch = match chain
|
||||
let Some(current_epoch) = chain
|
||||
.slot_clock
|
||||
.now()
|
||||
.map(|slot| slot.epoch(T::EthSpec::slots_per_epoch()))
|
||||
{
|
||||
Some(epoch) => epoch,
|
||||
None => continue, // we'll have to try again next time I suppose..
|
||||
else {
|
||||
continue;
|
||||
};
|
||||
|
||||
if current_epoch < deneb_fork_epoch {
|
||||
|
||||
@@ -547,9 +547,8 @@ impl<T: BeaconChainTypes> OverflowLRUCache<T> {
|
||||
.peek_lru()
|
||||
.map(|(key, value)| (*key, value.clone()));
|
||||
|
||||
let (lru_root, lru_pending_components) = match lru_entry {
|
||||
Some((r, p)) => (r, p),
|
||||
None => break,
|
||||
let Some((lru_root, lru_pending_components)) = lru_entry else {
|
||||
break;
|
||||
};
|
||||
|
||||
if lru_pending_components
|
||||
@@ -605,9 +604,8 @@ impl<T: BeaconChainTypes> OverflowLRUCache<T> {
|
||||
let delete_if_outdated = |cache: &OverflowLRUCache<T>,
|
||||
block_data: Option<BlockData>|
|
||||
-> Result<(), AvailabilityCheckError> {
|
||||
let block_data = match block_data {
|
||||
Some(block_data) => block_data,
|
||||
None => return Ok(()),
|
||||
let Some(block_data) = block_data else {
|
||||
return Ok(());
|
||||
};
|
||||
let not_in_store_keys = !cache.critical.read().store_keys.contains(&block_data.root);
|
||||
if not_in_store_keys {
|
||||
|
||||
@@ -99,9 +99,7 @@ impl<E: EthSpec> EarlyAttesterCache<E> {
|
||||
spec: &ChainSpec,
|
||||
) -> Result<Option<Attestation<E>>, Error> {
|
||||
let lock = self.item.read();
|
||||
let item = if let Some(item) = lock.as_ref() {
|
||||
item
|
||||
} else {
|
||||
let Some(item) = lock.as_ref() else {
|
||||
return Ok(None);
|
||||
};
|
||||
|
||||
|
||||
@@ -16,15 +16,14 @@ pub fn upgrade_to_v12<T: BeaconChainTypes>(
|
||||
let spec = db.get_chain_spec();
|
||||
|
||||
// Load a V5 op pool and transform it to V12.
|
||||
let PersistedOperationPoolV5 {
|
||||
let Some(PersistedOperationPoolV5 {
|
||||
attestations_v5,
|
||||
sync_contributions,
|
||||
attester_slashings_v5,
|
||||
proposer_slashings_v5,
|
||||
voluntary_exits_v5,
|
||||
} = if let Some(op_pool) = db.get_item(&OP_POOL_DB_KEY)? {
|
||||
op_pool
|
||||
} else {
|
||||
}) = db.get_item(&OP_POOL_DB_KEY)?
|
||||
else {
|
||||
debug!(log, "Nothing to do, no operation pool stored");
|
||||
return Ok(vec![]);
|
||||
};
|
||||
@@ -168,15 +167,14 @@ pub fn downgrade_from_v12<T: BeaconChainTypes>(
|
||||
log: Logger,
|
||||
) -> Result<Vec<KeyValueStoreOp>, Error> {
|
||||
// Load a V12 op pool and transform it to V5.
|
||||
let PersistedOperationPoolV12::<T::EthSpec> {
|
||||
let Some(PersistedOperationPoolV12::<T::EthSpec> {
|
||||
attestations,
|
||||
sync_contributions,
|
||||
attester_slashings,
|
||||
proposer_slashings,
|
||||
voluntary_exits,
|
||||
} = if let Some(op_pool_v12) = db.get_item(&OP_POOL_DB_KEY)? {
|
||||
op_pool_v12
|
||||
} else {
|
||||
}) = db.get_item(&OP_POOL_DB_KEY)?
|
||||
else {
|
||||
debug!(log, "Nothing to do, no operation pool stored");
|
||||
return Ok(vec![]);
|
||||
};
|
||||
|
||||
@@ -18,19 +18,14 @@ fn get_slot_clock<T: BeaconChainTypes>(
|
||||
log: &Logger,
|
||||
) -> Result<Option<T::SlotClock>, Error> {
|
||||
let spec = db.get_chain_spec();
|
||||
let genesis_block = if let Some(block) = db.get_blinded_block(&Hash256::zero())? {
|
||||
block
|
||||
} else {
|
||||
let Some(genesis_block) = db.get_blinded_block(&Hash256::zero())? else {
|
||||
error!(log, "Missing genesis block");
|
||||
return Ok(None);
|
||||
};
|
||||
let genesis_state =
|
||||
if let Some(state) = db.get_state(&genesis_block.state_root(), Some(Slot::new(0)))? {
|
||||
state
|
||||
} else {
|
||||
error!(log, "Missing genesis state"; "state_root" => ?genesis_block.state_root());
|
||||
return Ok(None);
|
||||
};
|
||||
let Some(genesis_state) = db.get_state(&genesis_block.state_root(), Some(Slot::new(0)))? else {
|
||||
error!(log, "Missing genesis state"; "state_root" => ?genesis_block.state_root());
|
||||
return Ok(None);
|
||||
};
|
||||
Ok(Some(T::SlotClock::new(
|
||||
spec.genesis_slot,
|
||||
Duration::from_secs(genesis_state.genesis_time()),
|
||||
@@ -43,15 +38,14 @@ pub fn upgrade_to_v14<T: BeaconChainTypes>(
|
||||
log: Logger,
|
||||
) -> Result<Vec<KeyValueStoreOp>, Error> {
|
||||
// Load a V12 op pool and transform it to V14.
|
||||
let PersistedOperationPoolV12::<T::EthSpec> {
|
||||
let Some(PersistedOperationPoolV12::<T::EthSpec> {
|
||||
attestations,
|
||||
sync_contributions,
|
||||
attester_slashings,
|
||||
proposer_slashings,
|
||||
voluntary_exits,
|
||||
} = if let Some(op_pool_v12) = db.get_item(&OP_POOL_DB_KEY)? {
|
||||
op_pool_v12
|
||||
} else {
|
||||
}) = db.get_item(&OP_POOL_DB_KEY)?
|
||||
else {
|
||||
debug!(log, "Nothing to do, no operation pool stored");
|
||||
return Ok(vec![]);
|
||||
};
|
||||
@@ -94,16 +88,15 @@ pub fn downgrade_from_v14<T: BeaconChainTypes>(
|
||||
}
|
||||
|
||||
// Load a V14 op pool and transform it to V12.
|
||||
let PersistedOperationPoolV14::<T::EthSpec> {
|
||||
let Some(PersistedOperationPoolV14::<T::EthSpec> {
|
||||
attestations,
|
||||
sync_contributions,
|
||||
attester_slashings,
|
||||
proposer_slashings,
|
||||
voluntary_exits,
|
||||
bls_to_execution_changes,
|
||||
} = if let Some(op_pool) = db.get_item(&OP_POOL_DB_KEY)? {
|
||||
op_pool
|
||||
} else {
|
||||
}) = db.get_item(&OP_POOL_DB_KEY)?
|
||||
else {
|
||||
debug!(log, "Nothing to do, no operation pool stored");
|
||||
return Ok(vec![]);
|
||||
};
|
||||
|
||||
@@ -11,16 +11,15 @@ pub fn upgrade_to_v15<T: BeaconChainTypes>(
|
||||
log: Logger,
|
||||
) -> Result<Vec<KeyValueStoreOp>, Error> {
|
||||
// Load a V14 op pool and transform it to V15.
|
||||
let PersistedOperationPoolV14::<T::EthSpec> {
|
||||
let Some(PersistedOperationPoolV14::<T::EthSpec> {
|
||||
attestations,
|
||||
sync_contributions,
|
||||
attester_slashings,
|
||||
proposer_slashings,
|
||||
voluntary_exits,
|
||||
bls_to_execution_changes,
|
||||
} = if let Some(op_pool_v14) = db.get_item(&OP_POOL_DB_KEY)? {
|
||||
op_pool_v14
|
||||
} else {
|
||||
}) = db.get_item(&OP_POOL_DB_KEY)?
|
||||
else {
|
||||
debug!(log, "Nothing to do, no operation pool stored");
|
||||
return Ok(vec![]);
|
||||
};
|
||||
@@ -43,7 +42,7 @@ pub fn downgrade_from_v15<T: BeaconChainTypes>(
|
||||
log: Logger,
|
||||
) -> Result<Vec<KeyValueStoreOp>, Error> {
|
||||
// Load a V15 op pool and transform it to V14.
|
||||
let PersistedOperationPoolV15::<T::EthSpec> {
|
||||
let Some(PersistedOperationPoolV15::<T::EthSpec> {
|
||||
attestations,
|
||||
sync_contributions,
|
||||
attester_slashings,
|
||||
@@ -51,9 +50,8 @@ pub fn downgrade_from_v15<T: BeaconChainTypes>(
|
||||
voluntary_exits,
|
||||
bls_to_execution_changes,
|
||||
capella_bls_change_broadcast_indices,
|
||||
} = if let Some(op_pool) = db.get_item(&OP_POOL_DB_KEY)? {
|
||||
op_pool
|
||||
} else {
|
||||
}) = db.get_item(&OP_POOL_DB_KEY)?
|
||||
else {
|
||||
debug!(log, "Nothing to do, no operation pool stored");
|
||||
return Ok(vec![]);
|
||||
};
|
||||
|
||||
@@ -17,19 +17,14 @@ fn get_slot_clock<T: BeaconChainTypes>(
|
||||
log: &Logger,
|
||||
) -> Result<Option<T::SlotClock>, Error> {
|
||||
let spec = db.get_chain_spec();
|
||||
let genesis_block = if let Some(block) = db.get_blinded_block(&Hash256::zero())? {
|
||||
block
|
||||
} else {
|
||||
let Some(genesis_block) = db.get_blinded_block(&Hash256::zero())? else {
|
||||
error!(log, "Missing genesis block");
|
||||
return Ok(None);
|
||||
};
|
||||
let genesis_state =
|
||||
if let Some(state) = db.get_state(&genesis_block.state_root(), Some(Slot::new(0)))? {
|
||||
state
|
||||
} else {
|
||||
error!(log, "Missing genesis state"; "state_root" => ?genesis_block.state_root());
|
||||
return Ok(None);
|
||||
};
|
||||
let Some(genesis_state) = db.get_state(&genesis_block.state_root(), Some(Slot::new(0)))? else {
|
||||
error!(log, "Missing genesis state"; "state_root" => ?genesis_block.state_root());
|
||||
return Ok(None);
|
||||
};
|
||||
Ok(Some(T::SlotClock::new(
|
||||
spec.genesis_slot,
|
||||
Duration::from_secs(genesis_state.genesis_time()),
|
||||
|
||||
@@ -113,14 +113,11 @@ async fn state_advance_timer<T: BeaconChainTypes>(
|
||||
let slot_duration = slot_clock.slot_duration();
|
||||
|
||||
loop {
|
||||
let duration_to_next_slot = match beacon_chain.slot_clock.duration_to_next_slot() {
|
||||
Some(duration) => duration,
|
||||
None => {
|
||||
error!(log, "Failed to read slot clock");
|
||||
// If we can't read the slot clock, just wait another slot.
|
||||
sleep(slot_duration).await;
|
||||
continue;
|
||||
}
|
||||
let Some(duration_to_next_slot) = beacon_chain.slot_clock.duration_to_next_slot() else {
|
||||
error!(log, "Failed to read slot clock");
|
||||
// If we can't read the slot clock, just wait another slot.
|
||||
sleep(slot_duration).await;
|
||||
continue;
|
||||
};
|
||||
|
||||
// Run the state advance 3/4 of the way through the slot (9s on mainnet).
|
||||
|
||||
@@ -1799,13 +1799,11 @@ impl<T: EthSpec> ExecutionLayer<T> {
|
||||
};
|
||||
}
|
||||
|
||||
let block = if let Some(block) = engine
|
||||
let Some(block) = engine
|
||||
.api
|
||||
.get_block_by_hash_with_txns::<T>(hash, fork)
|
||||
.await?
|
||||
{
|
||||
block
|
||||
} else {
|
||||
else {
|
||||
return Ok(None);
|
||||
};
|
||||
|
||||
|
||||
@@ -426,9 +426,7 @@ impl<T: EthSpec> ExecutionBlockGenerator<T> {
|
||||
}
|
||||
|
||||
pub fn new_payload(&mut self, payload: ExecutionPayload<T>) -> PayloadStatusV1 {
|
||||
let parent = if let Some(parent) = self.blocks.get(&payload.parent_hash()) {
|
||||
parent
|
||||
} else {
|
||||
let Some(parent) = self.blocks.get(&payload.parent_hash()) else {
|
||||
return PayloadStatusV1 {
|
||||
status: PayloadStatusV1Status::Syncing,
|
||||
latest_valid_hash: None,
|
||||
|
||||
@@ -30,9 +30,7 @@ pub fn sync_committee_duties<T: BeaconChainTypes>(
|
||||
request_indices: &[u64],
|
||||
chain: &BeaconChain<T>,
|
||||
) -> Result<SyncDuties, warp::reject::Rejection> {
|
||||
let altair_fork_epoch = if let Some(altair_fork_epoch) = chain.spec.altair_fork_epoch {
|
||||
altair_fork_epoch
|
||||
} else {
|
||||
let Some(altair_fork_epoch) = chain.spec.altair_fork_epoch else {
|
||||
// Empty response for networks with Altair disabled.
|
||||
return Ok(convert_to_response(vec![], false));
|
||||
};
|
||||
|
||||
@@ -135,9 +135,8 @@ impl<TSpec: EthSpec> Decoder for SSZSnappyInboundCodec<TSpec> {
|
||||
if self.protocol.versioned_protocol == SupportedProtocol::MetaDataV2 {
|
||||
return Ok(Some(InboundRequest::MetaData(MetadataRequest::new_v2())));
|
||||
}
|
||||
let length = match handle_length(&mut self.inner, &mut self.len, src)? {
|
||||
Some(len) => len,
|
||||
None => return Ok(None),
|
||||
let Some(length) = handle_length(&mut self.inner, &mut self.len, src)? else {
|
||||
return Ok(None);
|
||||
};
|
||||
|
||||
// Should not attempt to decode rpc chunks with `length > max_packet_size` or not within bounds of
|
||||
@@ -277,9 +276,8 @@ impl<TSpec: EthSpec> Decoder for SSZSnappyOutboundCodec<TSpec> {
|
||||
return Ok(None);
|
||||
}
|
||||
}
|
||||
let length = match handle_length(&mut self.inner, &mut self.len, src)? {
|
||||
Some(len) => len,
|
||||
None => return Ok(None),
|
||||
let Some(length) = handle_length(&mut self.inner, &mut self.len, src)? else {
|
||||
return Ok(None);
|
||||
};
|
||||
|
||||
// Should not attempt to decode rpc chunks with `length > max_packet_size` or not within bounds of
|
||||
@@ -324,9 +322,8 @@ impl<TSpec: EthSpec> OutboundCodec<OutboundRequest<TSpec>> for SSZSnappyOutbound
|
||||
&mut self,
|
||||
src: &mut BytesMut,
|
||||
) -> Result<Option<Self::CodecErrorType>, RPCError> {
|
||||
let length = match handle_length(&mut self.inner, &mut self.len, src)? {
|
||||
Some(len) => len,
|
||||
None => return Ok(None),
|
||||
let Some(length) = handle_length(&mut self.inner, &mut self.len, src)? else {
|
||||
return Ok(None);
|
||||
};
|
||||
|
||||
// Should not attempt to decode rpc chunks with `length > max_packet_size` or not within bounds of
|
||||
|
||||
@@ -286,9 +286,7 @@ where
|
||||
// wrong state a response will fail silently.
|
||||
fn send_response(&mut self, inbound_id: SubstreamId, response: RPCCodedResponse<TSpec>) {
|
||||
// check if the stream matching the response still exists
|
||||
let inbound_info = if let Some(info) = self.inbound_substreams.get_mut(&inbound_id) {
|
||||
info
|
||||
} else {
|
||||
let Some(inbound_info) = self.inbound_substreams.get_mut(&inbound_id) else {
|
||||
if !matches!(response, RPCCodedResponse::StreamTermination(..)) {
|
||||
// the stream is closed after sending the expected number of responses
|
||||
trace!(self.log, "Inbound stream has expired. Response not sent";
|
||||
@@ -296,7 +294,6 @@ where
|
||||
}
|
||||
return;
|
||||
};
|
||||
|
||||
// If the response we are sending is an error, report back for handling
|
||||
if let RPCCodedResponse::Error(ref code, ref reason) = response {
|
||||
self.events_out.push(Err(HandlerErr::Inbound {
|
||||
|
||||
@@ -205,9 +205,8 @@ impl GossipCache {
|
||||
GossipKind::LightClientFinalityUpdate => self.light_client_finality_update,
|
||||
GossipKind::LightClientOptimisticUpdate => self.light_client_optimistic_update,
|
||||
};
|
||||
let expire_timeout = match expire_timeout {
|
||||
Some(expire_timeout) => expire_timeout,
|
||||
None => return,
|
||||
let Some(expire_timeout) = expire_timeout else {
|
||||
return;
|
||||
};
|
||||
match self
|
||||
.topic_msgs
|
||||
|
||||
@@ -350,17 +350,14 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
|
||||
return;
|
||||
}
|
||||
};
|
||||
let bootstrap = match LightClientBootstrap::from_beacon_state(&mut beacon_state) {
|
||||
Ok(bootstrap) => bootstrap,
|
||||
Err(_) => {
|
||||
self.send_error_response(
|
||||
peer_id,
|
||||
RPCResponseErrorCode::ResourceUnavailable,
|
||||
"Bootstrap not available".into(),
|
||||
request_id,
|
||||
);
|
||||
return;
|
||||
}
|
||||
let Ok(bootstrap) = LightClientBootstrap::from_beacon_state(&mut beacon_state) else {
|
||||
self.send_error_response(
|
||||
peer_id,
|
||||
RPCResponseErrorCode::ResourceUnavailable,
|
||||
"Bootstrap not available".into(),
|
||||
request_id,
|
||||
);
|
||||
return;
|
||||
};
|
||||
self.send_response(
|
||||
peer_id,
|
||||
|
||||
@@ -115,34 +115,31 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
|
||||
duplicate_cache: DuplicateCache,
|
||||
) {
|
||||
// Check if the block is already being imported through another source
|
||||
let handle = match duplicate_cache.check_and_insert(block_root) {
|
||||
Some(handle) => handle,
|
||||
None => {
|
||||
debug!(
|
||||
self.log,
|
||||
"Gossip block is being processed";
|
||||
"action" => "sending rpc block to reprocessing queue",
|
||||
"block_root" => %block_root,
|
||||
);
|
||||
let Some(handle) = duplicate_cache.check_and_insert(block_root) else {
|
||||
debug!(
|
||||
self.log,
|
||||
"Gossip block is being processed";
|
||||
"action" => "sending rpc block to reprocessing queue",
|
||||
"block_root" => %block_root,
|
||||
);
|
||||
|
||||
// Send message to work reprocess queue to retry the block
|
||||
let (process_fn, ignore_fn) = self.clone().generate_rpc_beacon_block_fns(
|
||||
block_root,
|
||||
block,
|
||||
seen_timestamp,
|
||||
process_type,
|
||||
);
|
||||
let reprocess_msg = ReprocessQueueMessage::RpcBlock(QueuedRpcBlock {
|
||||
beacon_block_root: block_root,
|
||||
process_fn,
|
||||
ignore_fn,
|
||||
});
|
||||
// Send message to work reprocess queue to retry the block
|
||||
let (process_fn, ignore_fn) = self.clone().generate_rpc_beacon_block_fns(
|
||||
block_root,
|
||||
block,
|
||||
seen_timestamp,
|
||||
process_type,
|
||||
);
|
||||
let reprocess_msg = ReprocessQueueMessage::RpcBlock(QueuedRpcBlock {
|
||||
beacon_block_root: block_root,
|
||||
process_fn,
|
||||
ignore_fn,
|
||||
});
|
||||
|
||||
if reprocess_tx.try_send(reprocess_msg).is_err() {
|
||||
error!(self.log, "Failed to inform block import"; "source" => "rpc", "block_root" => %block_root)
|
||||
};
|
||||
return;
|
||||
}
|
||||
if reprocess_tx.try_send(reprocess_msg).is_err() {
|
||||
error!(self.log, "Failed to inform block import"; "source" => "rpc", "block_root" => %block_root)
|
||||
};
|
||||
return;
|
||||
};
|
||||
|
||||
// Returns `true` if the time now is after the 4s attestation deadline.
|
||||
|
||||
@@ -509,16 +509,13 @@ impl<T: BeaconChainTypes> BackFillSync<T> {
|
||||
return Ok(ProcessResult::Successful);
|
||||
}
|
||||
|
||||
let batch = match self.batches.get_mut(&batch_id) {
|
||||
Some(batch) => batch,
|
||||
None => {
|
||||
return self
|
||||
.fail_sync(BackFillError::InvalidSyncState(format!(
|
||||
"Trying to process a batch that does not exist: {}",
|
||||
batch_id
|
||||
)))
|
||||
.map(|_| ProcessResult::Successful);
|
||||
}
|
||||
let Some(batch) = self.batches.get_mut(&batch_id) else {
|
||||
return self
|
||||
.fail_sync(BackFillError::InvalidSyncState(format!(
|
||||
"Trying to process a batch that does not exist: {}",
|
||||
batch_id
|
||||
)))
|
||||
.map(|_| ProcessResult::Successful);
|
||||
};
|
||||
|
||||
// NOTE: We send empty batches to the processor in order to trigger the block processor
|
||||
@@ -909,9 +906,8 @@ impl<T: BeaconChainTypes> BackFillSync<T> {
|
||||
network: &mut SyncNetworkContext<T>,
|
||||
batch_id: BatchId,
|
||||
) -> Result<(), BackFillError> {
|
||||
let batch = match self.batches.get_mut(&batch_id) {
|
||||
Some(batch) => batch,
|
||||
None => return Ok(()),
|
||||
let Some(batch) = self.batches.get_mut(&batch_id) else {
|
||||
return Ok(());
|
||||
};
|
||||
|
||||
// Find a peer to request the batch
|
||||
|
||||
@@ -1015,15 +1015,12 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
|
||||
BlockProcessingResult::Ok(AvailabilityProcessingStatus::Imported(_))
|
||||
| BlockProcessingResult::Err(BlockError::BlockIsAlreadyKnown { .. }) => {
|
||||
// Check if the beacon processor is available
|
||||
let beacon_processor = match cx.beacon_processor_if_enabled() {
|
||||
Some(beacon_processor) => beacon_processor,
|
||||
None => {
|
||||
return trace!(
|
||||
self.log,
|
||||
"Dropping parent chain segment that was ready for processing.";
|
||||
parent_lookup
|
||||
);
|
||||
}
|
||||
let Some(beacon_processor) = cx.beacon_processor_if_enabled() else {
|
||||
return trace!(
|
||||
self.log,
|
||||
"Dropping parent chain segment that was ready for processing.";
|
||||
parent_lookup
|
||||
);
|
||||
};
|
||||
let (chain_hash, blocks, hashes, block_request) =
|
||||
parent_lookup.parts_for_processing();
|
||||
@@ -1195,11 +1192,8 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
|
||||
result: BatchProcessResult,
|
||||
cx: &SyncNetworkContext<T>,
|
||||
) {
|
||||
let request = match self.processing_parent_lookups.remove(&chain_hash) {
|
||||
Some((_hashes, request)) => request,
|
||||
None => {
|
||||
return debug!(self.log, "Chain process response for a parent lookup request that was not found"; "chain_hash" => %chain_hash, "result" => ?result)
|
||||
}
|
||||
let Some((_hashes, request)) = self.processing_parent_lookups.remove(&chain_hash) else {
|
||||
return debug!(self.log, "Chain process response for a parent lookup request that was not found"; "chain_hash" => %chain_hash, "result" => ?result);
|
||||
};
|
||||
|
||||
debug!(self.log, "Parent chain processed"; "chain_hash" => %chain_hash, "result" => ?result);
|
||||
|
||||
@@ -294,19 +294,15 @@ impl<T: BeaconChainTypes> SyncingChain<T> {
|
||||
return Ok(KeepChain);
|
||||
}
|
||||
|
||||
let beacon_processor = match network.beacon_processor_if_enabled() {
|
||||
Some(beacon_processor) => beacon_processor,
|
||||
None => return Ok(KeepChain),
|
||||
let Some(beacon_processor) = network.beacon_processor_if_enabled() else {
|
||||
return Ok(KeepChain);
|
||||
};
|
||||
|
||||
let batch = match self.batches.get_mut(&batch_id) {
|
||||
Some(batch) => batch,
|
||||
None => {
|
||||
return Err(RemoveChain::WrongChainState(format!(
|
||||
"Trying to process a batch that does not exist: {}",
|
||||
batch_id
|
||||
)));
|
||||
}
|
||||
let Some(batch) = self.batches.get_mut(&batch_id) else {
|
||||
return Err(RemoveChain::WrongChainState(format!(
|
||||
"Trying to process a batch that does not exist: {}",
|
||||
batch_id
|
||||
)));
|
||||
};
|
||||
|
||||
// NOTE: We send empty batches to the processor in order to trigger the block processor
|
||||
@@ -874,9 +870,8 @@ impl<T: BeaconChainTypes> SyncingChain<T> {
|
||||
network: &mut SyncNetworkContext<T>,
|
||||
batch_id: BatchId,
|
||||
) -> ProcessingResult {
|
||||
let batch = match self.batches.get_mut(&batch_id) {
|
||||
Some(batch) => batch,
|
||||
None => return Ok(KeepChain),
|
||||
let Some(batch) = self.batches.get_mut(&batch_id) else {
|
||||
return Ok(KeepChain);
|
||||
};
|
||||
|
||||
// Find a peer to request the batch
|
||||
|
||||
@@ -432,9 +432,8 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
|
||||
}
|
||||
|
||||
// Load the blinded block.
|
||||
let blinded_block = match self.get_blinded_block(block_root)? {
|
||||
Some(block) => block,
|
||||
None => return Ok(None),
|
||||
let Some(blinded_block) = self.get_blinded_block(block_root)? else {
|
||||
return Ok(None);
|
||||
};
|
||||
|
||||
// If the block is after the split point then we should have the full execution payload
|
||||
@@ -2053,12 +2052,9 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
|
||||
|
||||
/// Try to prune blobs, approximating the current epoch from the split slot.
|
||||
pub fn try_prune_most_blobs(&self, force: bool) -> Result<(), Error> {
|
||||
let deneb_fork_epoch = match self.spec.deneb_fork_epoch {
|
||||
Some(epoch) => epoch,
|
||||
None => {
|
||||
debug!(self.log, "Deneb fork is disabled");
|
||||
return Ok(());
|
||||
}
|
||||
let Some(deneb_fork_epoch) = self.spec.deneb_fork_epoch else {
|
||||
debug!(self.log, "Deneb fork is disabled");
|
||||
return Ok(());
|
||||
};
|
||||
// The current epoch is >= split_epoch + 2. It could be greater if the database is
|
||||
// configured to delay updating the split or finalization has ceased. In this instance we
|
||||
|
||||
@@ -17,9 +17,7 @@ where
|
||||
Cold: ItemStore<E>,
|
||||
{
|
||||
pub fn reconstruct_historic_states(self: &Arc<Self>) -> Result<(), Error> {
|
||||
let mut anchor = if let Some(anchor) = self.get_anchor_info() {
|
||||
anchor
|
||||
} else {
|
||||
let Some(mut anchor) = self.get_anchor_info() else {
|
||||
// Nothing to do, history is complete.
|
||||
return Ok(());
|
||||
};
|
||||
|
||||
@@ -16,12 +16,10 @@ pub fn spawn_timer<T: BeaconChainTypes>(
|
||||
let log = executor.log().clone();
|
||||
let timer_future = async move {
|
||||
loop {
|
||||
let duration_to_next_slot = match beacon_chain.slot_clock.duration_to_next_slot() {
|
||||
Some(duration) => duration,
|
||||
None => {
|
||||
warn!(log, "Unable to determine duration to next slot");
|
||||
return;
|
||||
}
|
||||
let Some(duration_to_next_slot) = beacon_chain.slot_clock.duration_to_next_slot()
|
||||
else {
|
||||
warn!(log, "Unable to determine duration to next slot");
|
||||
return;
|
||||
};
|
||||
|
||||
sleep(duration_to_next_slot).await;
|
||||
|
||||
Reference in New Issue
Block a user