Fix flaky unknown block test

beacon_node/network/src/network_beacon_processor/gossip_methods.rs

@@ -1005,7 +1005,7 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
                         "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<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
                 // 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<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
                         "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<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
                             header_or_bid.clone(),
                             block_root,
                             publish_blobs,
+                            EnvelopeSource::Gossip,
                         )
                         .await;
                     self_clone
@@ -1942,7 +1948,7 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
 
         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<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
 
     /// 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::<T::EthSpec>(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<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
             .is_err()
         {
             error!(
-                source = "gossip",
+                source = source.as_ref(),
                 ?block_root,
                 "Failed to inform block import"
             )

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<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
         header_or_bid: PartialHeaderOrBid<T::EthSpec>,
         block_root: Hash256,
         publish_blobs: bool,
+        source: EnvelopeSource,
     ) {
         if self.chain.config.disable_get_blobs {
             return;
@@ -954,13 +956,14 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
         .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!(

beacon_node/network/src/network_beacon_processor/sync_methods.rs

@@ -209,6 +209,7 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
                         header_or_bid,
                         block_root,
                         publish_blobs,
+                        EnvelopeSource::Rpc,
                     )
                     .await;
                 } else {

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::<E>();
+    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) {