Avoid computing columns from EL blobs if block has already been imported (#6816)

Squashed commit of the following:

commit 0df2384
Author: Jimmy Chen <[email protected]>
Date:   Tue Jan 21 16:41:30 2025 +1100

    Downgrade a `warn` log to `debug` and update handling.

commit b3584ac
Author: Jimmy Chen <[email protected]>
Date:   Fri Jan 17 22:39:16 2025 +1100

    Avoid computing columns from EL blobs if block has already been imported.

Author: jimmygchen

beacon_node/beacon_chain/src/fetch_blobs.rs

@@ -163,6 +163,20 @@ pub async fn fetch_and_process_engine_blobs<T: BeaconChainTypes>(
             return Ok(None);
         }
 
+        if chain
+            .canonical_head
+            .fork_choice_read_lock()
+            .contains_block(&block_root)
+        {
+            // Avoid computing columns if block has already been imported.
+            debug!(
+                log,
+                "Ignoring EL blobs response";
+                "info" => "block has already been imported",
+            );
+            return Ok(None);
+        }
+
         let data_columns_receiver = spawn_compute_and_publish_data_columns_task(
             &chain,
             block.clone(),
@@ -248,18 +262,21 @@ fn spawn_compute_and_publish_data_columns_task<T: BeaconChainTypes>(
                 }
             };
 
-            if let Err(e) = data_columns_sender.send(all_data_columns.clone()) {
-                error!(log, "Failed to send computed data columns"; "error" => ?e);
+            if data_columns_sender.send(all_data_columns.clone()).is_err() {
+                // Data column receiver have been dropped - block may have already been imported.
+                // This race condition exists because gossip columns may arrive and trigger block
+                // import during the computation. Here we just drop the computed columns.
+                debug!(
+                    log,
+                    "Failed to send computed data columns";
+                );
+                return;
             };
 
-            // Check indices from cache before sending the columns, to make sure we don't
-            // publish components already seen on gossip.
-            let is_supernode = chain_cloned.data_availability_checker.is_supernode();
-
             // At the moment non supernodes are not required to publish any columns.
             // TODO(das): we could experiment with having full nodes publish their custodied
             // columns here.
-            if !is_supernode {
+            if !chain_cloned.data_availability_checker.is_supernode() {
                 return;
             }