Verify getBlobsV2 response and avoid reprocessing imported data columns (#7493)

#7461 and partly #6439.

Desired behaviour after receiving `engine_getBlobs` response:

1. Gossip verify the blobs and proofs, but don't mark them as observed yet. This is because not all blobs are published immediately (due to staggered publishing). If we mark them as observed and not publish them, we could end up blocking the gossip propagation.
2. Blobs are marked as observed _either_ when:
* They are received from gossip and forwarded to the network .
* They are published by the node.

Current behaviour:
- ❗ We only gossip verify `engine_getBlobsV1` responses, but not `engine_getBlobsV2` responses (PeerDAS).
- ❗ After importing EL blobs AND before they're published, if the same blobs arrive via gossip, they will get re-processed, which may result in a re-import.


  1. Perform gossip verification on data columns computed from EL `getBlobsV2` response. We currently only do this for `getBlobsV1` to prevent importing blobs with invalid proofs into the `DataAvailabilityChecker`, this should be done on V2 responses too.
2. Add additional gossip verification to make sure we don't re-process a ~~blob~~ or data column that was imported via the EL `getBlobs` but not yet "seen" on the gossip network. If an "unobserved" gossip blob is found in the availability cache, then we know it has passed verification so we can immediately propagate the `ACCEPT` result and forward it to the network, but without re-processing it.

**UPDATE:** I've left blobs out for the second change mentioned above, as the likelihood and impact is very slow and we haven't seen it enough, but under PeerDAS this issue is a regular occurrence and we do see the same block getting imported many times.

beacon_node/beacon_chain/src/data_column_verification.rs

@@ -129,6 +129,10 @@ pub enum GossipDataColumnError {
         slot: Slot,
         index: ColumnIndex,
     },
+    /// A column has already been processed from non-gossip source and have not yet been seen on
+    /// the gossip network.
+    /// This column should be accepted and forwarded over gossip.
+    PriorKnownUnpublished,
     /// Data column index must be between 0 and `NUMBER_OF_COLUMNS` (exclusive).
     ///
     /// ## Peer scoring
@@ -181,6 +185,16 @@ pub struct GossipVerifiedDataColumn<T: BeaconChainTypes, O: ObservationStrategy
     _phantom: PhantomData<O>,
 }
 
+impl<T: BeaconChainTypes, O: ObservationStrategy> Clone for GossipVerifiedDataColumn<T, O> {
+    fn clone(&self) -> Self {
+        Self {
+            block_root: self.block_root,
+            data_column: self.data_column.clone(),
+            _phantom: PhantomData,
+        }
+    }
+}
+
 impl<T: BeaconChainTypes, O: ObservationStrategy> GossipVerifiedDataColumn<T, O> {
     pub fn new(
         column_sidecar: Arc<DataColumnSidecar<T::EthSpec>>,
@@ -200,6 +214,16 @@ impl<T: BeaconChainTypes, O: ObservationStrategy> GossipVerifiedDataColumn<T, O>
         )
     }
 
+    /// Create a `GossipVerifiedDataColumn` from `DataColumnSidecar` for testing ONLY.
+    #[cfg(test)]
+    pub(crate) fn __new_for_testing(column_sidecar: Arc<DataColumnSidecar<T::EthSpec>>) -> Self {
+        Self {
+            block_root: column_sidecar.block_root(),
+            data_column: KzgVerifiedDataColumn::__new_for_testing(column_sidecar),
+            _phantom: Default::default(),
+        }
+    }
+
     pub fn as_data_column(&self) -> &DataColumnSidecar<T::EthSpec> {
         self.data_column.as_data_column()
     }
@@ -243,11 +267,9 @@ impl<E: EthSpec> KzgVerifiedDataColumn<E> {
         verify_kzg_for_data_column(data_column, kzg)
     }
 
-    /// Create a `KzgVerifiedDataColumn` from `data_column` that are already KZG verified.
-    ///
-    /// This should be used with caution, as used incorrectly it could result in KZG verification
-    /// being skipped and invalid data_columns being deemed valid.
-    pub fn from_verified(data_column: Arc<DataColumnSidecar<E>>) -> Self {
+    /// Create a `KzgVerifiedDataColumn` from `DataColumnSidecar` for testing ONLY.
+    #[cfg(test)]
+    pub(crate) fn __new_for_testing(data_column: Arc<DataColumnSidecar<E>>) -> Self {
         Self { data: data_column }
     }
 
@@ -444,6 +466,23 @@ pub fn validate_data_column_sidecar_for_gossip<T: BeaconChainTypes, O: Observati
     verify_sidecar_not_from_future_slot(chain, column_slot)?;
     verify_slot_greater_than_latest_finalized_slot(chain, column_slot)?;
     verify_is_first_sidecar(chain, &data_column)?;
+
+    // Check if the data column is already in the DA checker cache. This happens when data columns
+    // are made available through the `engine_getBlobs` method.  If it exists in the cache, we know
+    // it has already passed the gossip checks, even though this particular instance hasn't been
+    // seen / published on the gossip network yet (passed the `verify_is_first_sidecar` check above).
+    // In this case, we should accept it for gossip propagation.
+    if chain
+        .data_availability_checker
+        .is_data_column_cached(&data_column.block_root(), &data_column)
+    {
+        // Observe this data column so we don't process it again.
+        if O::observe() {
+            observe_gossip_data_column(&data_column, chain)?;
+        }
+        return Err(GossipDataColumnError::PriorKnownUnpublished);
+    }
+
     verify_column_inclusion_proof(&data_column)?;
     let parent_block = verify_parent_block_and_finalized_descendant(data_column.clone(), chain)?;
     verify_slot_higher_than_parent(&parent_block, column_slot)?;
@@ -463,7 +502,7 @@ pub fn validate_data_column_sidecar_for_gossip<T: BeaconChainTypes, O: Observati
         .map_err(|e| GossipDataColumnError::BeaconChainError(Box::new(e.into())))?;
 
     if O::observe() {
-        observe_gossip_data_column(&kzg_verified_data_column.data, chain)?;
+        observe_gossip_data_column(&data_column, chain)?;
     }
 
     Ok(GossipVerifiedDataColumn {