From 21b643318049991cae6015779a88abd45420cedc Mon Sep 17 00:00:00 2001 From: Michael Sproul Date: Tue, 8 Sep 2026 13:02:56 +1000 Subject: [PATCH 1/3] Notify reprocessing queue after importing engine blobs If the EL blob fetch wins the block import race, queued attestations never receive the import notification and attestation_to_unknown_block_processed_after_gossip_block can time out. Reuse the column import notification helper so this path also releases waiting work, including payload attestations after Gloas. --- .../network/src/network_beacon_processor/gossip_methods.rs | 2 +- beacon_node/network/src/network_beacon_processor/mod.rs | 3 ++- 2 files changed, 3 insertions(+), 2 deletions(-) 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..b76c6b92b93 100644 --- a/beacon_node/network/src/network_beacon_processor/gossip_methods.rs +++ b/beacon_node/network/src/network_beacon_processor/gossip_methods.rs @@ -4088,7 +4088,7 @@ 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) { + pub(super) fn notify_import_after_column(&self, slot: Slot, block_root: Hash256) { if self .chain .spec diff --git a/beacon_node/network/src/network_beacon_processor/mod.rs b/beacon_node/network/src/network_beacon_processor/mod.rs index b6b383b4eea..755e401b461 100644 --- a/beacon_node/network/src/network_beacon_processor/mod.rs +++ b/beacon_node/network/src/network_beacon_processor/mod.rs @@ -954,13 +954,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); } AvailabilityProcessingStatus::MissingComponents(_, _) => { debug!( From a57cfe1c1fc39b329a864d89ad0e01b01f738f79 Mon Sep 17 00:00:00 2001 From: Michael Sproul Date: Tue, 8 Sep 2026 13:19:14 +1000 Subject: [PATCH 2/3] Test attestation reprocessing after engine blob import Process the Fulu block without its columns before explicitly fetching blobs from the EL. Assert that the block becomes head and its queued attestation is reprocessed and imported. This deterministically fails when the EL import notification is missing. --- .../src/network_beacon_processor/tests.rs | 54 +++++++++++++++++++ 1 file changed, 54 insertions(+) diff --git a/beacon_node/network/src/network_beacon_processor/tests.rs b/beacon_node/network/src/network_beacon_processor/tests.rs index dda34b8e653..1a86eb9fc1e 100644 --- a/beacon_node/network/src/network_beacon_processor/tests.rs +++ b/beacon_node/network/src/network_beacon_processor/tests.rs @@ -1403,6 +1403,60 @@ 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::{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, + ) + .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) { From 43e756cf9b8be666d429d0035c05795f381e5b0e Mon Sep 17 00:00:00 2001 From: Michael Sproul Date: Wed, 9 Sep 2026 13:46:33 +1000 Subject: [PATCH 3/3] Thread envelope source --- .../gossip_methods.rs | 33 +++++++++++++------ .../src/network_beacon_processor/mod.rs | 4 ++- .../network_beacon_processor/sync_methods.rs | 1 + .../src/network_beacon_processor/tests.rs | 2 ++ 4 files changed, 29 insertions(+), 11 deletions(-) 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 b76c6b92b93..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. - pub(super) 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 755e401b461..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; @@ -961,7 +963,7 @@ impl NetworkBeaconProcessor { "Block components retrieved from EL" ); self.chain.recompute_head_at_current_slot().await; - self.notify_import_after_column(slot, block_root); + 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 1a86eb9fc1e..9052a5ff40b 100644 --- a/beacon_node/network/src/network_beacon_processor/tests.rs +++ b/beacon_node/network/src/network_beacon_processor/tests.rs @@ -1406,6 +1406,7 @@ async fn attestation_to_unknown_block_processed_after_rpc_block() { #[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; @@ -1446,6 +1447,7 @@ async fn attestation_to_unknown_block_processed_after_engine_blobs() { PartialHeaderOrBid::try_from_block(rig.next_block.as_ref()).unwrap(), block_root, false, + EnvelopeSource::Gossip, ) .await; assert_eq!(rig.head_root(), block_root);