Cleaner logic for gossip subscriptions for new forks (sigp#4030)

Cleaner resolution for sigp#4006

We are currently subscribing to core topics of new forks way before the actual fork since we had just a single `CORE_TOPICS` array. This PR separates the core topics for every fork and subscribes to only required topics based on the current fork.
Also adds logic for subscribing to the core topics of a new fork only 2 slots before the fork happens.

2 slots is to give enough time for the gossip meshes to form.

Currently doesn't add logic to remove topics from older forks in new forks. For e.g. in the coupled 4844 world, we had to remove the `BeaconBlock` topic in favour of `BeaconBlocksAndBlobsSidecar` at the 4844 fork. It should be easy enough to add though. Not adding it because I'm assuming that  sigp#4019 will get merged before this PR and we won't require any deletion logic. Happy to add it regardless though.

beacon_node/lighthouse_network/src/types/mod.rs

@@ -17,6 +17,6 @@ pub use pubsub::{PubsubMessage, SignedBeaconBlockAndBlobsSidecar, SnappyTransfor
 pub use subnet::{Subnet, SubnetDiscovery};
 pub use sync_state::{BackFillState, SyncState};
 pub use topics::{
-    subnet_from_topic_hash, GossipEncoding, GossipKind, GossipTopic, CORE_TOPICS,
-    LIGHT_CLIENT_GOSSIP_TOPICS,
+    core_topics_to_subscribe, fork_core_topics, subnet_from_topic_hash, GossipEncoding, GossipKind,
+    GossipTopic, LIGHT_CLIENT_GOSSIP_TOPICS,
 };

beacon_node/network/src/service.rs

@@ -19,7 +19,7 @@ use lighthouse_network::{
     Context, PeerAction, PeerRequestId, PubsubMessage, ReportSource, Request, Response, Subnet,
 };
 use lighthouse_network::{
-    types::{GossipEncoding, GossipTopic},
+    types::{core_topics_to_subscribe, GossipEncoding, GossipTopic},
     MessageId, NetworkEvent, NetworkGlobals, PeerId,
 };
 use slog::{crit, debug, error, info, o, trace, warn};
@@ -445,7 +445,7 @@ impl<T: BeaconChainTypes> NetworkService<T> {
                             let fork_version = self.beacon_chain.spec.fork_version_for_name(fork_name);
                             let fork_digest = ChainSpec::compute_fork_digest(fork_version, self.beacon_chain.genesis_validators_root);
                             info!(self.log, "Subscribing to new fork topics");
-                            self.libp2p.subscribe_new_fork_topics(fork_digest);
+                            self.libp2p.subscribe_new_fork_topics(fork_name, fork_digest);
                             self.next_fork_subscriptions = Box::pin(None.into());
                         }
                         else {
@@ -685,7 +685,7 @@ impl<T: BeaconChainTypes> NetworkService<T> {
                 }
 
                 let mut subscribed_topics: Vec<GossipTopic> = vec![];
-                for topic_kind in lighthouse_network::types::CORE_TOPICS.iter() {
+                for topic_kind in core_topics_to_subscribe(self.fork_context.current_fork()) {
                     for fork_digest in self.required_gossip_fork_digests() {
                         let topic = GossipTopic::new(
                             topic_kind.clone(),