gnosis/lighthouse
1
0
Fork 0
mirror of https://github.com/sigp/lighthouse.git synced 2026-05-30 12:47:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add allow reprocess flag

Browse Source
This commit is contained in:
Eitan Seri-Levi
2026-05-24 10:15:13 +03:00
parent 623c8f4617
commit 255773834b
4 changed files with 35 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
beacon_node/network/src/network_beacon_processor/gossip_methods.rs
View File

@@ -649,6 +649,7 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
subnet_id: DataColumnSubnetId, subnet_id: DataColumnSubnetId,
column_sidecar: Arc<DataColumnSidecar<T::EthSpec>>, column_sidecar: Arc<DataColumnSidecar<T::EthSpec>>,
seen_duration: Duration, seen_duration: Duration,
allow_reprocess: bool,
) { ) {
let slot = column_sidecar.slot(); let slot = column_sidecar.slot();
let block_root = column_sidecar.block_root(); let block_root = column_sidecar.block_root();
@@ -738,33 +739,37 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
MessageAcceptance::Ignore, MessageAcceptance::Ignore,
); );
// Queue the column for reprocessing when the block arrives. if allow_reprocess {
let processor = self.clone(); // Queue the column for reprocessing when the block arrives.
let reprocess_msg = let processor = self.clone();
ReprocessQueueMessage::UnknownBlockDataColumn(QueuedGossipDataColumn { let reprocess_msg = ReprocessQueueMessage::UnknownBlockDataColumn(
beacon_block_root: unknown_block_root, QueuedGossipDataColumn {
process_fn: Box::new(move || { beacon_block_root: unknown_block_root,
let _ = processor.send_gossip_data_column_sidecar( process_fn: Box::new(move || {
message_id, let _ = processor.send_gossip_data_column_sidecar(
peer_id, message_id,
subnet_id, peer_id,
column_sidecar, subnet_id,
seen_duration, column_sidecar,
); seen_duration,
}), false, // Do not reprocess this message again.
}); );
if self }),
.beacon_processor_send },
.try_send(WorkEvent {
drop_during_sync: false,
work: Work::Reprocess(reprocess_msg),
})
.is_err()
{
debug!(
%unknown_block_root,
"Failed to queue data column for reprocessing"
); );
if self
.beacon_processor_send
.try_send(WorkEvent {
drop_during_sync: false,
work: Work::Reprocess(reprocess_msg),
})
.is_err()
{
debug!(
%unknown_block_root,
"Failed to queue data column for reprocessing"
);
}
} }
} }
GossipDataColumnError::InvalidVariant GossipDataColumnError::InvalidVariant

2
beacon_node/network/src/network_beacon_processor/mod.rs
View File

@@ -231,6 +231,7 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
subnet_id: DataColumnSubnetId, subnet_id: DataColumnSubnetId,
column_sidecar: Arc<DataColumnSidecar<T::EthSpec>>, column_sidecar: Arc<DataColumnSidecar<T::EthSpec>>,
seen_timestamp: Duration, seen_timestamp: Duration,
allow_reprocess: bool,
) -> Result<(), Error<T::EthSpec>> { ) -> Result<(), Error<T::EthSpec>> {
let processor = self.clone(); let processor = self.clone();
let process_fn = async move { let process_fn = async move {
@@ -241,6 +242,7 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
subnet_id, subnet_id,
column_sidecar, column_sidecar,
seen_timestamp, seen_timestamp,
allow_reprocess,
) )
.await .await
}; };

1
beacon_node/network/src/network_beacon_processor/tests.rs
View File

@@ -435,6 +435,7 @@ impl TestRig {
DataColumnSubnetId::from_column_index(*data_column.index(), &self.chain.spec), DataColumnSubnetId::from_column_index(*data_column.index(), &self.chain.spec),
data_column.clone(), data_column.clone(),
Duration::from_secs(0), Duration::from_secs(0),
true,
) )
.unwrap(); .unwrap();
} }

1
beacon_node/network/src/router.rs
View File

@@ -435,6 +435,7 @@ impl<T: BeaconChainTypes> Router<T> {
subnet_id, subnet_id,
column_sidecar, column_sidecar,
seen_timestamp, seen_timestamp,
true,
), ),
) )
} }