Remove CGC from data_availability checker (#7033)

- Part of https://github.com/sigp/lighthouse/issues/6767

Validator custody makes the CGC and set of sampling columns dynamic. Right now this information is stored twice:
- in the data availability checker
- in the network globals

If that state becomes dynamic we must make sure it is in sync updating it twice, or guarding it behind a mutex. However, I noted that we don't really have to keep the CGC inside the data availability checker. All consumers can actually read it from the network globals, and we can update `make_available` to read the expected count of data columns from the block.

beacon_node/network/src/network_beacon_processor/gossip_methods.rs

@@ -1257,7 +1257,10 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
         let verification_result = self
             .chain
             .clone()
-            .verify_block_for_gossip(block.clone())
+            .verify_block_for_gossip(
+                block.clone(),
+                self.network_globals.custody_columns_count() as usize,
+            )
             .await;
 
         if verification_result.is_ok() {

beacon_node/network/src/network_beacon_processor/mod.rs

@@ -927,9 +927,14 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
         block_root: Hash256,
         publish_blobs: bool,
     ) {
+        let is_supernode = self.network_globals.is_supernode();
+
         let self_cloned = self.clone();
         let publish_fn = move |blobs_or_data_column| {
-            if publish_blobs {
+            // 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 publish_blobs && is_supernode {
                 match blobs_or_data_column {
                     BlobsOrDataColumns::Blobs(blobs) => {
                         self_cloned.publish_blobs_gradually(blobs, block_root);
@@ -1004,6 +1009,11 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
         self: &Arc<Self>,
         block_root: Hash256,
     ) -> Option<AvailabilityProcessingStatus> {
+        // Only supernodes attempt reconstruction
+        if !self.network_globals.is_supernode() {
+            return None;
+        }
+
         let result = self.chain.reconstruct_data_columns(block_root).await;
         match result {
             Ok(Some((availability_processing_status, data_columns_to_publish))) => {

beacon_node/network/src/sync/block_sidecar_coupling.rs

@@ -257,8 +257,14 @@ impl<E: EthSpec> RangeBlockComponentsRequest<E> {
                     ));
                 }
 
-                RpcBlock::new_with_custody_columns(Some(block_root), block, custody_columns, spec)
-                    .map_err(|e| format!("{e:?}"))?
+                RpcBlock::new_with_custody_columns(
+                    Some(block_root),
+                    block,
+                    custody_columns,
+                    expects_custody_columns.len(),
+                    spec,
+                )
+                .map_err(|e| format!("{e:?}"))?
             } else {
                 RpcBlock::new_without_blobs(Some(block_root), block)
             });

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

@@ -1203,8 +1203,12 @@ impl TestRig {
             payload_verification_status: PayloadVerificationStatus::Verified,
             is_valid_merge_transition_block: false,
         };
-        let executed_block =
-            AvailabilityPendingExecutedBlock::new(block, import_data, payload_verification_outcome);
+        let executed_block = AvailabilityPendingExecutedBlock::new(
+            block,
+            import_data,
+            payload_verification_outcome,
+            self.network_globals.custody_columns_count() as usize,
+        );
         match self
             .harness
             .chain

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

@@ -449,7 +449,15 @@ fn build_rpc_block(
             RpcBlock::new(None, block, Some(blobs.clone())).unwrap()
         }
         Some(DataSidecars::DataColumns(columns)) => {
-            RpcBlock::new_with_custody_columns(None, block, columns.clone(), spec).unwrap()
+            RpcBlock::new_with_custody_columns(
+                None,
+                block,
+                columns.clone(),
+                // TODO(das): Assumes CGC = max value. Change if we want to do more complex tests
+                columns.len(),
+                spec,
+            )
+            .unwrap()
         }
         None => RpcBlock::new_without_blobs(None, block),
     }