Subscribe to PeerDAS topics on Fulu fork (#6849)

`TODO(das)` now that PeerDAS is scheduled in a hard fork we can subscribe to its topics on the fork activation. In current stable we subscribe to PeerDAS topics as soon as the node starts if PeerDAS is scheduled.

This PR adds another todo to unsubscribe to blob topics at the fork. This other PR included solution for that, but I can include it in a separate PR
- https://github.com/sigp/lighthouse/pull/5899/files


  Include PeerDAS topics as part of Fulu fork in `fork_core_topics`.

beacon_node/lighthouse_network/src/service/mod.rs

@@ -708,11 +708,17 @@ impl<E: EthSpec> Network<E> {
         }
 
         // Subscribe to core topics for the new fork
-        for kind in fork_core_topics::<E>(&new_fork, &self.fork_context.spec) {
+        for kind in fork_core_topics::<E>(
+            &new_fork,
+            &self.fork_context.spec,
+            &self.network_globals.as_topic_config(),
+        ) {
             let topic = GossipTopic::new(kind, GossipEncoding::default(), new_fork_digest);
             self.subscribe(topic);
         }
 
+        // TODO(das): unsubscribe from blob topics at the Fulu fork
+
         // Register the new topics for metrics
         let topics_to_keep_metrics_for = attestation_sync_committee_topics::<E>()
             .map(|gossip_kind| {

beacon_node/lighthouse_network/src/types/globals.rs

@@ -1,4 +1,5 @@
 //! A collection of variables that are accessible outside of the network thread itself.
+use super::TopicConfig;
 use crate::peer_manager::peerdb::PeerDB;
 use crate::rpc::{MetaData, MetaDataV3};
 use crate::types::{BackFillState, SyncState};
@@ -183,6 +184,14 @@ impl<E: EthSpec> NetworkGlobals<E> {
             .collect::<Vec<_>>()
     }
 
+    /// Returns the TopicConfig to compute the set of Gossip topics for a given fork
+    pub fn as_topic_config(&self) -> TopicConfig {
+        TopicConfig {
+            subscribe_all_data_column_subnets: self.config.subscribe_all_data_column_subnets,
+            sampling_subnets: &self.sampling_subnets,
+        }
+    }
+
     /// TESTING ONLY. Build a dummy NetworkGlobals instance.
     pub fn new_test_globals(
         trusted_peers: Vec<PeerId>,

beacon_node/lighthouse_network/src/types/mod.rs

@@ -17,6 +17,6 @@ pub use subnet::{Subnet, SubnetDiscovery};
 pub use sync_state::{BackFillState, SyncState};
 pub use topics::{
     attestation_sync_committee_topics, core_topics_to_subscribe, fork_core_topics,
-    subnet_from_topic_hash, GossipEncoding, GossipKind, GossipTopic, ALTAIR_CORE_TOPICS,
-    BASE_CORE_TOPICS, CAPELLA_CORE_TOPICS, LIGHT_CLIENT_GOSSIP_TOPICS,
+    subnet_from_topic_hash, GossipEncoding, GossipKind, GossipTopic, TopicConfig,
+    ALTAIR_CORE_TOPICS, BASE_CORE_TOPICS, CAPELLA_CORE_TOPICS, LIGHT_CLIENT_GOSSIP_TOPICS,
 };

beacon_node/lighthouse_network/src/types/topics.rs

@@ -1,5 +1,6 @@
 use gossipsub::{IdentTopic as Topic, TopicHash};
 use serde::{Deserialize, Serialize};
+use std::collections::HashSet;
 use strum::AsRefStr;
 use types::{ChainSpec, DataColumnSubnetId, EthSpec, ForkName, SubnetId, SyncSubnetId, Unsigned};
 
@@ -41,8 +42,18 @@ pub const LIGHT_CLIENT_GOSSIP_TOPICS: [GossipKind; 2] = [
     GossipKind::LightClientOptimisticUpdate,
 ];
 
+#[derive(Debug)]
+pub struct TopicConfig<'a> {
+    pub subscribe_all_data_column_subnets: bool,
+    pub sampling_subnets: &'a HashSet<DataColumnSubnetId>,
+}
+
 /// Returns the core topics associated with each fork that are new to the previous fork
-pub fn fork_core_topics<E: EthSpec>(fork_name: &ForkName, spec: &ChainSpec) -> Vec<GossipKind> {
+pub fn fork_core_topics<E: EthSpec>(
+    fork_name: &ForkName,
+    spec: &ChainSpec,
+    topic_config: &TopicConfig,
+) -> Vec<GossipKind> {
     match fork_name {
         ForkName::Base => BASE_CORE_TOPICS.to_vec(),
         ForkName::Altair => ALTAIR_CORE_TOPICS.to_vec(),
@@ -64,7 +75,21 @@ pub fn fork_core_topics<E: EthSpec>(fork_name: &ForkName, spec: &ChainSpec) -> V
             }
             electra_blob_topics
         }
-        ForkName::Fulu => vec![],
+        ForkName::Fulu => {
+            let mut topics = vec![];
+            if topic_config.subscribe_all_data_column_subnets {
+                for column_subnet in 0..spec.data_column_sidecar_subnet_count {
+                    topics.push(GossipKind::DataColumnSidecar(DataColumnSubnetId::new(
+                        column_subnet,
+                    )));
+                }
+            } else {
+                for column_subnet in topic_config.sampling_subnets {
+                    topics.push(GossipKind::DataColumnSidecar(*column_subnet));
+                }
+            }
+            topics
+        }
     }
 }
 
@@ -84,10 +109,11 @@ pub fn attestation_sync_committee_topics<E: EthSpec>() -> impl Iterator<Item = G
 pub fn core_topics_to_subscribe<E: EthSpec>(
     mut current_fork: ForkName,
     spec: &ChainSpec,
+    topic_config: &TopicConfig,
 ) -> Vec<GossipKind> {
-    let mut topics = fork_core_topics::<E>(&current_fork, spec);
+    let mut topics = fork_core_topics::<E>(&current_fork, spec, topic_config);
     while let Some(previous_fork) = current_fork.previous_fork() {
-        let previous_fork_topics = fork_core_topics::<E>(&previous_fork, spec);
+        let previous_fork_topics = fork_core_topics::<E>(&previous_fork, spec, topic_config);
         topics.extend(previous_fork_topics);
         current_fork = previous_fork;
     }
@@ -475,8 +501,15 @@ mod tests {
         type E = MainnetEthSpec;
         let spec = E::default_spec();
         let mut all_topics = Vec::new();
-        let mut electra_core_topics = fork_core_topics::<E>(&ForkName::Electra, &spec);
-        let mut deneb_core_topics = fork_core_topics::<E>(&ForkName::Deneb, &spec);
+        let topic_config = TopicConfig {
+            subscribe_all_data_column_subnets: false,
+            sampling_subnets: &HashSet::from_iter([1, 2].map(DataColumnSubnetId::new)),
+        };
+        let mut fulu_core_topics = fork_core_topics::<E>(&ForkName::Fulu, &spec, &topic_config);
+        let mut electra_core_topics =
+            fork_core_topics::<E>(&ForkName::Electra, &spec, &topic_config);
+        let mut deneb_core_topics = fork_core_topics::<E>(&ForkName::Deneb, &spec, &topic_config);
+        all_topics.append(&mut fulu_core_topics);
         all_topics.append(&mut electra_core_topics);
         all_topics.append(&mut deneb_core_topics);
         all_topics.extend(CAPELLA_CORE_TOPICS);
@@ -484,7 +517,7 @@ mod tests {
         all_topics.extend(BASE_CORE_TOPICS);
 
         let latest_fork = *ForkName::list_all().last().unwrap();
-        let core_topics = core_topics_to_subscribe::<E>(latest_fork, &spec);
+        let core_topics = core_topics_to_subscribe::<E>(latest_fork, &spec, &topic_config);
         // Need to check all the topics exist in an order independent manner
         for topic in all_topics {
             assert!(core_topics.contains(&topic));