Only mark block lookups as pending if block is importing from gossip (#8112)

- PR https://github.com/sigp/lighthouse/pull/8045 introduced a regression of how lookup sync interacts with the da_checker.

Now in unstable block import from the HTTP API also insert the block in the da_checker while the block is being execution verified. If lookup sync finds the block in the da_checker in `NotValidated` state it expects a `GossipBlockProcessResult` message sometime later. That message is only sent after block import in gossip.

I confirmed in our node's logs for 4/4 cases of stuck lookups are caused by this sequence of events:
- Receive block through API, insert into da_checker in fn process_block in put_pre_execution_block
- Create lookup and leave in AwaitingDownload(block in processing cache) state
- Block from HTTP API finishes importing
- Lookup is left stuck

Closes https://github.com/sigp/lighthouse/issues/8104


  - https://github.com/sigp/lighthouse/pull/8110 was my initial solution attempt but we can't send the `GossipBlockProcessResult` event from the `http_api` crate without adding new channels, which seems messy.

For a given node it's rare that a lookup is created at the same time that a block is being published. This PR solves https://github.com/sigp/lighthouse/issues/8104 by allowing lookup sync to import the block twice in that case.


Co-Authored-By: dapplion <35266934+dapplion@users.noreply.github.com>

beacon_node/beacon_chain/src/data_availability_checker/overflow_lru_cache.rs

@@ -19,13 +19,14 @@ use tracing::{Span, debug, debug_span};
 use types::beacon_block_body::KzgCommitments;
 use types::blob_sidecar::BlobIdentifier;
 use types::{
-    BlobSidecar, ChainSpec, ColumnIndex, DataColumnSidecar, DataColumnSidecarList, Epoch, EthSpec,
-    Hash256, RuntimeFixedVector, RuntimeVariableList, SignedBeaconBlock,
+    BlobSidecar, BlockImportSource, ChainSpec, ColumnIndex, DataColumnSidecar,
+    DataColumnSidecarList, Epoch, EthSpec, Hash256, RuntimeFixedVector, RuntimeVariableList,
+    SignedBeaconBlock,
 };
 
 #[derive(Clone)]
 pub enum CachedBlock<E: EthSpec> {
-    PreExecution(Arc<SignedBeaconBlock<E>>),
+    PreExecution(Arc<SignedBeaconBlock<E>>, BlockImportSource),
     Executed(Box<DietAvailabilityPendingExecutedBlock<E>>),
 }
 
@@ -42,7 +43,7 @@ impl<E: EthSpec> CachedBlock<E> {
 
     fn as_block(&self) -> &SignedBeaconBlock<E> {
         match self {
-            CachedBlock::PreExecution(b) => b,
+            CachedBlock::PreExecution(b, _) => b,
             CachedBlock::Executed(b) => b.as_block(),
         }
     }
@@ -135,9 +136,13 @@ impl<E: EthSpec> PendingComponents<E> {
 
     /// Inserts a pre-execution block into the cache.
     /// This does NOT override an existing executed block.
-    pub fn insert_pre_execution_block(&mut self, block: Arc<SignedBeaconBlock<E>>) {
+    pub fn insert_pre_execution_block(
+        &mut self,
+        block: Arc<SignedBeaconBlock<E>>,
+        source: BlockImportSource,
+    ) {
         if self.block.is_none() {
-            self.block = Some(CachedBlock::PreExecution(block))
+            self.block = Some(CachedBlock::PreExecution(block, source))
         }
     }
 
@@ -433,7 +438,9 @@ impl<T: BeaconChainTypes> DataAvailabilityCheckerInner<T> {
             .peek(block_root)
             .and_then(|pending_components| {
                 pending_components.block.as_ref().map(|block| match block {
-                    CachedBlock::PreExecution(b) => BlockProcessStatus::NotValidated(b.clone()),
+                    CachedBlock::PreExecution(b, source) => {
+                        BlockProcessStatus::NotValidated(b.clone(), *source)
+                    }
                     CachedBlock::Executed(b) => {
                         BlockProcessStatus::ExecutionValidated(b.block_cloned())
                     }
@@ -693,11 +700,12 @@ impl<T: BeaconChainTypes> DataAvailabilityCheckerInner<T> {
         &self,
         block_root: Hash256,
         block: Arc<SignedBeaconBlock<T::EthSpec>>,
+        source: BlockImportSource,
     ) -> Result<(), AvailabilityCheckError> {
         let epoch = block.epoch();
         let pending_components =
             self.update_or_insert_pending_components(block_root, epoch, |pending_components| {
-                pending_components.insert_pre_execution_block(block);
+                pending_components.insert_pre_execution_block(block, source);
                 Ok(())
             })?;
 
@@ -718,7 +726,7 @@ impl<T: BeaconChainTypes> DataAvailabilityCheckerInner<T> {
     /// This does NOT remove an existing executed block.
     pub fn remove_pre_execution_block(&self, block_root: &Hash256) {
         // The read lock is immediately dropped so we can safely remove the block from the cache.
-        if let Some(BlockProcessStatus::NotValidated(_)) = self.get_cached_block(block_root) {
+        if let Some(BlockProcessStatus::NotValidated(_, _)) = self.get_cached_block(block_root) {
             self.critical.write().pop(block_root);
         }
     }
@@ -1459,9 +1467,13 @@ mod pending_components_tests {
         let mut pending_component = <PendingComponents<E>>::empty(block_root, max_len);
 
         let pre_execution_block = Arc::new(pre_execution_block);
-        pending_component.insert_pre_execution_block(pre_execution_block.clone());
+        pending_component
+            .insert_pre_execution_block(pre_execution_block.clone(), BlockImportSource::Gossip);
         assert!(
-            matches!(pending_component.block, Some(CachedBlock::PreExecution(_))),
+            matches!(
+                pending_component.block,
+                Some(CachedBlock::PreExecution(_, _))
+            ),
             "pre execution block inserted"
         );
 
@@ -1471,7 +1483,8 @@ mod pending_components_tests {
             "executed block inserted"
         );
 
-        pending_component.insert_pre_execution_block(pre_execution_block);
+        pending_component
+            .insert_pre_execution_block(pre_execution_block, BlockImportSource::Gossip);
         assert!(
             matches!(pending_component.block, Some(CachedBlock::Executed(_))),
             "executed block should remain"