Blocklookup data inconsistencies (#3677)

## Issue Addressed
Closes #3649 

## Proposed Changes

Add a regression test for the data inconsistency, catching the problem in 31e88c5533 [here](https://github.com/sigp/lighthouse/actions/runs/3379894044/jobs/5612044797#step:6:2043).
When a chain is sent for processing, move it to a separate collection and now the test works, yay!

## Additional Info

na

beacon_node/network/src/sync/block_lookups/tests.rs

@@ -259,7 +259,7 @@ fn test_single_block_lookup_becomes_parent_request() {
     assert_eq!(bl.single_block_lookups.len(), 0);
     rig.expect_parent_request();
     rig.expect_empty_network();
-    assert_eq!(bl.parent_queue.len(), 1);
+    assert_eq!(bl.parent_lookups.len(), 1);
 }
 
 #[test]
@@ -287,7 +287,7 @@ fn test_parent_lookup_happy_path() {
         was_non_empty: true,
     };
     bl.parent_chain_processed(chain_hash, process_result, &mut cx);
-    assert_eq!(bl.parent_queue.len(), 0);
+    assert_eq!(bl.parent_lookups.len(), 0);
 }
 
 #[test]
@@ -324,7 +324,7 @@ fn test_parent_lookup_wrong_response() {
         was_non_empty: true,
     };
     bl.parent_chain_processed(chain_hash, process_result, &mut cx);
-    assert_eq!(bl.parent_queue.len(), 0);
+    assert_eq!(bl.parent_lookups.len(), 0);
 }
 
 #[test]
@@ -356,7 +356,7 @@ fn test_parent_lookup_empty_response() {
         was_non_empty: true,
     };
     bl.parent_chain_processed(chain_hash, process_result, &mut cx);
-    assert_eq!(bl.parent_queue.len(), 0);
+    assert_eq!(bl.parent_lookups.len(), 0);
 }
 
 #[test]
@@ -387,7 +387,7 @@ fn test_parent_lookup_rpc_failure() {
         was_non_empty: true,
     };
     bl.parent_chain_processed(chain_hash, process_result, &mut cx);
-    assert_eq!(bl.parent_queue.len(), 0);
+    assert_eq!(bl.parent_lookups.len(), 0);
 }
 
 #[test]
@@ -419,11 +419,11 @@ fn test_parent_lookup_too_many_attempts() {
             }
         }
         if i < parent_lookup::PARENT_FAIL_TOLERANCE {
-            assert_eq!(bl.parent_queue[0].failed_attempts(), dbg!(i));
+            assert_eq!(bl.parent_lookups[0].failed_attempts(), dbg!(i));
         }
     }
 
-    assert_eq!(bl.parent_queue.len(), 0);
+    assert_eq!(bl.parent_lookups.len(), 0);
 }
 
 #[test]
@@ -450,11 +450,11 @@ fn test_parent_lookup_too_many_download_attempts_no_blacklist() {
             rig.expect_penalty();
         }
         if i < parent_lookup::PARENT_FAIL_TOLERANCE {
-            assert_eq!(bl.parent_queue[0].failed_attempts(), dbg!(i));
+            assert_eq!(bl.parent_lookups[0].failed_attempts(), dbg!(i));
         }
     }
 
-    assert_eq!(bl.parent_queue.len(), 0);
+    assert_eq!(bl.parent_lookups.len(), 0);
     assert!(!bl.failed_chains.contains(&block_hash));
     assert!(!bl.failed_chains.contains(&parent.canonical_root()));
 }
@@ -491,7 +491,7 @@ fn test_parent_lookup_too_many_processing_attempts_must_blacklist() {
     }
 
     assert!(bl.failed_chains.contains(&block_hash));
-    assert_eq!(bl.parent_queue.len(), 0);
+    assert_eq!(bl.parent_lookups.len(), 0);
 }
 
 #[test]
@@ -545,7 +545,7 @@ fn test_parent_lookup_disconnection() {
         &mut cx,
     );
     bl.peer_disconnected(&peer_id, &mut cx);
-    assert!(bl.parent_queue.is_empty());
+    assert!(bl.parent_lookups.is_empty());
 }
 
 #[test]
@@ -598,5 +598,78 @@ fn test_parent_lookup_ignored_response() {
     // Return an Ignored result. The request should be dropped
     bl.parent_block_processed(chain_hash, BlockProcessResult::Ignored, &mut cx);
     rig.expect_empty_network();
-    assert_eq!(bl.parent_queue.len(), 0);
+    assert_eq!(bl.parent_lookups.len(), 0);
+}
+
+/// This is a regression test.
+#[test]
+fn test_same_chain_race_condition() {
+    let (mut bl, mut cx, mut rig) = TestRig::test_setup(Some(Level::Debug));
+
+    #[track_caller]
+    fn parent_lookups_consistency(bl: &BlockLookups<T>) {
+        let hashes: Vec<_> = bl
+            .parent_lookups
+            .iter()
+            .map(|req| req.chain_hash())
+            .collect();
+        let expected = hashes.len();
+        assert_eq!(
+            expected,
+            hashes
+                .into_iter()
+                .collect::<std::collections::HashSet<_>>()
+                .len(),
+            "duplicated chain hashes in parent queue"
+        )
+    }
+    // if we use one or two blocks it will match on the hash or the parent hash, so make a longer
+    // chain.
+    let depth = 4;
+    let mut blocks = Vec::<Arc<SignedBeaconBlock<E>>>::with_capacity(depth);
+    while blocks.len() < depth {
+        let parent = blocks
+            .last()
+            .map(|b| b.canonical_root())
+            .unwrap_or_else(Hash256::random);
+        let block = Arc::new(rig.block_with_parent(parent));
+        blocks.push(block);
+    }
+
+    let peer_id = PeerId::random();
+    let trigger_block = blocks.pop().unwrap();
+    let chain_hash = trigger_block.canonical_root();
+    bl.search_parent(chain_hash, trigger_block.clone(), peer_id, &mut cx);
+
+    for (i, block) in blocks.into_iter().rev().enumerate() {
+        let id = rig.expect_parent_request();
+        // the block
+        bl.parent_lookup_response(id, peer_id, Some(block.clone()), D, &mut cx);
+        // the stream termination
+        bl.parent_lookup_response(id, peer_id, None, D, &mut cx);
+        // the processing request
+        rig.expect_block_process();
+        // the processing result
+        if i + 2 == depth {
+            // one block was removed
+            bl.parent_block_processed(chain_hash, BlockError::BlockIsAlreadyKnown.into(), &mut cx)
+        } else {
+            bl.parent_block_processed(chain_hash, BlockError::ParentUnknown(block).into(), &mut cx)
+        }
+        parent_lookups_consistency(&bl)
+    }
+
+    // Processing succeeds, now the rest of the chain should be sent for processing.
+    rig.expect_parent_chain_process();
+
+    // Try to get this block again while the chain is being processed. We should not request it again.
+    let peer_id = PeerId::random();
+    bl.search_parent(chain_hash, trigger_block, peer_id, &mut cx);
+    parent_lookups_consistency(&bl);
+
+    let process_result = BatchProcessResult::Success {
+        was_non_empty: true,
+    };
+    bl.parent_chain_processed(chain_hash, process_result, &mut cx);
+    assert_eq!(bl.parent_lookups.len(), 0);
 }