diff --git a/beacon_node/network/src/network_beacon_processor/gossip_methods.rs b/beacon_node/network/src/network_beacon_processor/gossip_methods.rs index c78f6ae913e..f23ab123b86 100644 --- a/beacon_node/network/src/network_beacon_processor/gossip_methods.rs +++ b/beacon_node/network/src/network_beacon_processor/gossip_methods.rs @@ -1005,7 +1005,7 @@ impl NetworkBeaconProcessor { "Gossipsub data column processed, imported fully available block" ); self.chain.recompute_head_at_current_slot().await; - self.notify_import_after_column(slot, block_root); + self.notify_import_after_column(slot, block_root, EnvelopeSource::Gossip); metrics::set_gauge( &metrics::BEACON_BLOB_DELAY_FULL_VERIFICATION, @@ -1130,8 +1130,13 @@ impl NetworkBeaconProcessor { // We want to publish immediately when this finishes let publish_blobs = true; let header = PartialHeaderOrBid::PartialHeader(header.into_header()); - self.fetch_engine_blobs_and_publish_full(header.clone(), block_root, publish_blobs) - .await; + self.fetch_engine_blobs_and_publish_full( + header.clone(), + block_root, + publish_blobs, + EnvelopeSource::Gossip, + ) + .await; self.publish_partial_data_columns(header, block_root).await; } } @@ -1449,7 +1454,7 @@ impl NetworkBeaconProcessor { "Data column from partial processed, imported fully available block" ); self.chain.recompute_head_at_current_slot().await; - self.notify_import_after_column(*slot, *block_root); + self.notify_import_after_column(*slot, *block_root, EnvelopeSource::Gossip); metrics::set_gauge( &metrics::BEACON_BLOB_DELAY_FULL_VERIFICATION, @@ -1917,6 +1922,7 @@ impl NetworkBeaconProcessor { header_or_bid.clone(), block_root, publish_blobs, + EnvelopeSource::Gossip, ) .await; self_clone @@ -1942,7 +1948,7 @@ impl NetworkBeaconProcessor { match &result { Ok(AvailabilityProcessingStatus::Imported(_, block_root)) => { - self.notify_block_imported(*block_root); + self.notify_block_imported(*block_root, EnvelopeSource::Gossip); debug!( ?block_root, @@ -4088,21 +4094,28 @@ impl NetworkBeaconProcessor { /// Inform the reprocess queue that a fully available block (or its payload envelope, post-gloas) /// has been imported, so any attestations waiting on it can be released. - fn notify_import_after_column(&self, slot: Slot, block_root: Hash256) { + /// `source` identifies the import path for logging. + pub(super) fn notify_import_after_column( + &self, + slot: Slot, + block_root: Hash256, + source: EnvelopeSource, + ) { if self .chain .spec .fork_name_at_slot::(slot) .gloas_enabled() { - self.notify_payload_envelope_imported(block_root, EnvelopeSource::Gossip); + self.notify_payload_envelope_imported(block_root, source); } else { - self.notify_block_imported(block_root); + self.notify_block_imported(block_root, source); } } /// Inform the reprocess queue that `block_root` has been imported as a full block. - fn notify_block_imported(&self, block_root: Hash256) { + /// `source` identifies the import path for logging. + fn notify_block_imported(&self, block_root: Hash256, source: EnvelopeSource) { if self .beacon_processor_send .try_send(WorkEvent { @@ -4112,7 +4125,7 @@ impl NetworkBeaconProcessor { .is_err() { error!( - source = "gossip", + source = source.as_ref(), ?block_root, "Failed to inform block import" ) diff --git a/beacon_node/network/src/network_beacon_processor/mod.rs b/beacon_node/network/src/network_beacon_processor/mod.rs index b6b383b4eea..8f301d7e3a9 100644 --- a/beacon_node/network/src/network_beacon_processor/mod.rs +++ b/beacon_node/network/src/network_beacon_processor/mod.rs @@ -9,6 +9,7 @@ use beacon_chain::fetch_blobs::{ FetchEngineBlobError, PartialHeaderOrBid, fetch_and_process_engine_blobs, }; use beacon_chain::partial_data_column_assembler::AssemblyColumn; +use beacon_chain::payload_envelope_verification::EnvelopeSource; use beacon_chain::test_utils::{BeaconChainHarness, EphemeralHarnessType}; use beacon_chain::{AvailabilityProcessingStatus, BeaconChain, BeaconChainTypes, BlockError}; use beacon_processor::{ @@ -928,6 +929,7 @@ impl NetworkBeaconProcessor { header_or_bid: PartialHeaderOrBid, block_root: Hash256, publish_blobs: bool, + source: EnvelopeSource, ) { if self.chain.config.disable_get_blobs { return; @@ -954,13 +956,14 @@ impl NetworkBeaconProcessor { .await { Ok(Some(availability)) => match availability { - AvailabilityProcessingStatus::Imported(..) => { + AvailabilityProcessingStatus::Imported(slot, block_root) => { debug!( result = "imported block and custody columns", %block_root, "Block components retrieved from EL" ); self.chain.recompute_head_at_current_slot().await; + self.notify_import_after_column(slot, block_root, source); } AvailabilityProcessingStatus::MissingComponents(_, _) => { debug!( diff --git a/beacon_node/network/src/network_beacon_processor/sync_methods.rs b/beacon_node/network/src/network_beacon_processor/sync_methods.rs index 582d8aeafdf..c9d40e7b8b7 100644 --- a/beacon_node/network/src/network_beacon_processor/sync_methods.rs +++ b/beacon_node/network/src/network_beacon_processor/sync_methods.rs @@ -209,6 +209,7 @@ impl NetworkBeaconProcessor { header_or_bid, block_root, publish_blobs, + EnvelopeSource::Rpc, ) .await; } else { diff --git a/beacon_node/network/src/network_beacon_processor/tests.rs b/beacon_node/network/src/network_beacon_processor/tests.rs index dda34b8e653..9052a5ff40b 100644 --- a/beacon_node/network/src/network_beacon_processor/tests.rs +++ b/beacon_node/network/src/network_beacon_processor/tests.rs @@ -1403,6 +1403,62 @@ async fn attestation_to_unknown_block_processed_after_rpc_block() { attestation_to_unknown_block_processed(BlockImportMethod::Rpc).await } +#[tokio::test] +async fn attestation_to_unknown_block_processed_after_engine_blobs() { + use beacon_chain::fetch_blobs::PartialHeaderOrBid; + use beacon_chain::payload_envelope_verification::EnvelopeSource; + use beacon_chain::{AvailabilityProcessingStatus, NotifyExecutionLayer}; + use types::block::BlockImportSource; + + // This block/data availability ordering applies only to Fulu. + let spec = test_spec::(); + if spec.fulu_fork_epoch.is_none() || spec.gloas_fork_epoch.is_some() { + return; + } + + let mut rig = TestRig::new(SMALL_CHAIN).await; + let initial_attns = rig.chain.naive_aggregation_pool.read().num_items(); + rig.enqueue_next_block_unaggregated_attestation(); + rig.assert_event_journal_completes(&[WorkType::GossipAttestation]) + .await; + assert_eq!( + rig.chain.naive_aggregation_pool.read().num_items(), + initial_attns + ); + + // Process the block directly, without starting the network processor's concurrent EL fetch. + // Withhold the columns so the explicit EL fetch below must finish importing the block. + let block_root = rig.next_block.canonical_root(); + let result = rig + .chain + .process_block( + block_root, + LookupBlock::new(rig.next_block.clone()), + NotifyExecutionLayer::Yes, + BlockImportSource::Gossip, + || Ok(()), + ) + .await + .unwrap(); + assert_matches!(result, AvailabilityProcessingStatus::MissingComponents(..)); + + rig.network_beacon_processor + .fetch_engine_blobs_and_publish_full( + PartialHeaderOrBid::try_from_block(rig.next_block.as_ref()).unwrap(), + block_root, + false, + EnvelopeSource::Gossip, + ) + .await; + assert_eq!(rig.head_root(), block_root); + rig.assert_event_journal_contains_ordered(&[WorkType::UnknownBlockAttestation]) + .await; + assert_eq!( + rig.chain.naive_aggregation_pool.read().num_items(), + initial_attns + 1 + ); +} + /// Ensure that attestations that reference an unknown block get properly re-queued and /// re-processed upon importing the block. async fn aggregate_attestation_to_unknown_block(import_method: BlockImportMethod) {