De-duplicate stateful variables in NetworkGlobals

beacon_node/http_api/src/test_utils.rs

@@ -8,15 +8,13 @@ use beacon_processor::{
 };
 use directory::DEFAULT_ROOT_DIR;
 use eth2::{BeaconNodeHttpClient, Timeouts};
-use lighthouse_network::rpc::methods::MetaDataV3;
 use lighthouse_network::{
     discv5::enr::CombinedKey,
     libp2p::swarm::{
         behaviour::{ConnectionEstablished, FromSwarm},
         ConnectionId, NetworkBehaviour,
     },
-    rpc::methods::{MetaData, MetaDataV2},
+    types::SyncState,
-    types::{EnrAttestationBitfield, EnrSyncCommitteeBitfield, SyncState},
     ConnectedPoint, Enr, NetworkConfig, NetworkGlobals, PeerId, PeerManager,
 };
 use network::{NetworkReceivers, NetworkSenders};
@@ -139,28 +137,11 @@ pub async fn create_api_server_with_config<T: BeaconChainTypes>(
 
     let (network_senders, network_receivers) = NetworkSenders::new();
 
-    // Default metadata
-    let meta_data = if chain.spec.is_peer_das_scheduled() {
-        MetaData::V3(MetaDataV3 {
-            seq_number: SEQ_NUMBER,
-            attnets: EnrAttestationBitfield::<T::EthSpec>::default(),
-            syncnets: EnrSyncCommitteeBitfield::<T::EthSpec>::default(),
-            custody_group_count: chain.spec.custody_requirement,
-        })
-    } else {
-        MetaData::V2(MetaDataV2 {
-            seq_number: SEQ_NUMBER,
-            attnets: EnrAttestationBitfield::<T::EthSpec>::default(),
-            syncnets: EnrSyncCommitteeBitfield::<T::EthSpec>::default(),
-        })
-    };
-
     let enr_key = CombinedKey::generate_secp256k1();
     let enr = Enr::builder().build(&enr_key).unwrap();
     let network_config = Arc::new(NetworkConfig::default());
     let network_globals = Arc::new(NetworkGlobals::new(
         enr.clone(),
-        meta_data,
         vec![],
         false,
         network_config,

beacon_node/lighthouse_network/src/discovery/mod.rs

@@ -209,7 +209,7 @@ impl<E: EthSpec> Discovery<E> {
             None => String::from(""),
         };
 
-        let local_enr = network_globals.local_enr.read().clone();
+        let local_enr = network_globals.local_enr();
         let local_node_id = local_enr.node_id();
 
         info!(
@@ -420,7 +420,7 @@ impl<E: EthSpec> Discovery<E> {
             .map_err(|e| format!("{:?}", e))?;
 
         // replace the global version
-        *self.network_globals.local_enr.write() = self.discv5.local_enr();
+        self.network_globals.set_enr(self.discv5.local_enr());
         // persist modified enr to disk
         enr::save_enr_to_disk(Path::new(&self.enr_dir), &self.local_enr());
         Ok(true)
@@ -456,7 +456,7 @@ impl<E: EthSpec> Discovery<E> {
             .map_err(|e| format!("{:?}", e))?;
 
         // replace the global version
-        *self.network_globals.local_enr.write() = self.discv5.local_enr();
+        self.network_globals.set_enr(self.discv5.local_enr());
         // persist modified enr to disk
         enr::save_enr_to_disk(Path::new(&self.enr_dir), &self.local_enr());
         Ok(true)
@@ -472,7 +472,7 @@ impl<E: EthSpec> Discovery<E> {
             // persist modified enr to disk
             enr::save_enr_to_disk(Path::new(&self.enr_dir), &self.local_enr());
         }
-        *self.network_globals.local_enr.write() = self.discv5.local_enr();
+        self.network_globals.set_enr(self.discv5.local_enr());
         Ok(())
     }
 
@@ -553,7 +553,7 @@ impl<E: EthSpec> Discovery<E> {
         }
 
         // replace the global version
-        *self.network_globals.local_enr.write() = self.discv5.local_enr();
+        self.network_globals.set_enr(self.discv5.local_enr());
 
         // persist modified enr to disk
         enr::save_enr_to_disk(Path::new(&self.enr_dir), &self.local_enr());
@@ -588,7 +588,7 @@ impl<E: EthSpec> Discovery<E> {
             });
 
         // replace the global version with discovery version
-        *self.network_globals.local_enr.write() = self.discv5.local_enr();
+        self.network_globals.set_enr(self.discv5.local_enr());
 
         // persist modified enr to disk
         enr::save_enr_to_disk(Path::new(&self.enr_dir), &self.local_enr());
@@ -1060,7 +1060,7 @@ impl<E: EthSpec> NetworkBehaviour for Discovery<E> {
                             let enr = self.discv5.local_enr();
                             enr::save_enr_to_disk(Path::new(&self.enr_dir), &enr);
                             // update  network globals
-                            *self.network_globals.local_enr.write() = enr;
+                            self.network_globals.set_enr(enr);
                             // A new UDP socket has been detected.
                             // NOTE: We assume libp2p itself can keep track of IP changes and we do
                             // not inform it about IP changes found via discovery.
@@ -1194,7 +1194,6 @@ impl<E: EthSpec> Discovery<E> {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::rpc::methods::{MetaData, MetaDataV2};
     use libp2p::identity::secp256k1;
     use types::{BitVector, MinimalEthSpec, SubnetId};
 
@@ -1208,18 +1207,7 @@ mod tests {
         let config = Arc::new(config);
         let enr_key: CombinedKey = CombinedKey::from_secp256k1(&keypair);
         let enr: Enr = build_enr::<E>(&enr_key, &config, &EnrForkId::default(), &spec).unwrap();
-        let globals = NetworkGlobals::new(
+        let globals = NetworkGlobals::new(enr, vec![], false, config.clone(), spec.clone());
-            enr,
-            MetaData::V2(MetaDataV2 {
-                seq_number: 0,
-                attnets: Default::default(),
-                syncnets: Default::default(),
-            }),
-            vec![],
-            false,
-            config.clone(),
-            spec.clone(),
-        );
         let keypair = keypair.into();
         Discovery::new(keypair, &config, Arc::new(globals), &spec)
             .await

beacon_node/lighthouse_network/src/rpc/mod.rs

@@ -288,6 +288,9 @@ impl<Id: ReqId, E: EthSpec> RPC<Id, E> {
         name = "libp2p_rpc",
         skip_all
     )]
+    // TODO: Check who uses this, and pass seq_number as a parameter
+    // Introduced here, but I don't follow why we can't read it from network globals
+    // https://github.com/sigp/lighthouse/pull/6400
     pub fn ping(&mut self, peer_id: PeerId, id: Id) {
         let ping = Ping {
             data: self.seq_number,

beacon_node/lighthouse_network/src/service/mod.rs

@@ -34,7 +34,6 @@ use libp2p::upnp::tokio::Behaviour as Upnp;
 use libp2p::{identify, PeerId, SwarmBuilder};
 use logging::crit;
 use std::num::{NonZeroU8, NonZeroUsize};
-use std::path::PathBuf;
 use std::pin::Pin;
 use std::sync::Arc;
 use std::time::Duration;
@@ -153,7 +152,6 @@ pub struct Network<E: EthSpec> {
     // lookups for every gossipsub message send.
     enr_fork_id: EnrForkId,
     /// Directory where metadata is stored.
-    network_dir: PathBuf,
     fork_context: Arc<ForkContext>,
     /// Gossipsub score parameters.
     score_settings: PeerScoreSettings<E>,
@@ -199,15 +197,8 @@ impl<E: EthSpec> Network<E> {
         )?;
 
         // Construct the metadata
-        let custody_group_count = ctx.chain_spec.is_peer_das_scheduled().then(|| {
-            ctx.chain_spec
-                .custody_group_count(config.subscribe_all_data_column_subnets)
-        });
-        let meta_data = utils::load_or_build_metadata(&config.network_dir, custody_group_count);
-        let seq_number = *meta_data.seq_number();
         let globals = NetworkGlobals::new(
             enr,
-            meta_data,
             trusted_peers,
             config.disable_peer_scoring,
             config.clone(),
@@ -369,6 +360,7 @@ impl<E: EthSpec> Network<E> {
             ttfb_timeout: ctx.chain_spec.ttfb_timeout(),
             resp_timeout: ctx.chain_spec.resp_timeout(),
         };
+        let seq_number = *network_globals.local_metadata().seq_number();
         let eth2_rpc = RPC::new(
             ctx.fork_context.clone(),
             config.enable_light_client_server,
@@ -505,7 +497,6 @@ impl<E: EthSpec> Network<E> {
             swarm,
             network_globals,
             enr_fork_id,
-            network_dir: config.network_dir.clone(),
             fork_context: ctx.fork_context,
             score_settings,
             update_gossipsub_scores,
@@ -1257,8 +1248,11 @@ impl<E: EthSpec> Network<E> {
         if let Err(e) = self.discovery_mut().update_enr_bitfield(subnet_id, value) {
             crit!(error = e, "Could not update ENR bitfield");
         }
-        // update the local meta data which informs our peers of the update during PINGS
+
-        self.update_metadata_bitfields();
+        // TODO: Can we deprecate this for a single source of truth?
+        let metadata = self.network_globals.local_metadata();
+        self.eth2_rpc_mut()
+            .update_seq_number(*metadata.seq_number());
     }
 
     /// Attempts to discover new peers for a given subnet. The `min_ttl` gives the time at which we
@@ -1338,37 +1332,6 @@ impl<E: EthSpec> Network<E> {
 
     /* Private internal functions */
 
-    /// Updates the current meta data of the node to match the local ENR.
-    #[instrument(parent = None,
-        level = "trace",
-        fields(service = "libp2p"),
-        name = "libp2p",
-        skip_all
-    )]
-    fn update_metadata_bitfields(&mut self) {
-        let local_attnets = self
-            .discovery_mut()
-            .local_enr()
-            .attestation_bitfield::<E>()
-            .expect("Local discovery must have attestation bitfield");
-
-        let local_syncnets = self
-            .discovery_mut()
-            .local_enr()
-            .sync_committee_bitfield::<E>()
-            .expect("Local discovery must have sync committee bitfield");
-
-        // write lock scope
-        self.network_globals
-            .update_metadata_bitfields(local_attnets, local_syncnets);
-
-        let metadata = self.network_globals.local_metadata();
-        self.eth2_rpc_mut()
-            .update_seq_number(*metadata.seq_number());
-        // Save the updated metadata to disk
-        utils::save_metadata_to_disk(&self.network_dir, metadata);
-    }
-
     /// Sends a Ping request to the peer.
     #[instrument(parent = None,
         level = "trace",

beacon_node/lighthouse_network/src/types/globals.rs

@@ -1,7 +1,8 @@
 //! A collection of variables that are accessible outside of the network thread itself.
 use super::TopicConfig;
+use crate::discovery::enr::Eth2Enr;
 use crate::peer_manager::peerdb::PeerDB;
-use crate::rpc::{MetaData, MetaDataV3};
+use crate::rpc::{MetaData, MetaDataV2, MetaDataV3};
 use crate::types::{BackFillState, SyncState};
 use crate::{Client, Enr, EnrExt, GossipTopic, Multiaddr, NetworkConfig, PeerId};
 use parking_lot::RwLock;
@@ -10,19 +11,17 @@ use std::sync::Arc;
 use types::data_column_custody_group::{
     compute_columns_for_custody_group, compute_subnets_from_custody_group, get_custody_groups,
 };
-use types::{BitVector, ChainSpec, ColumnIndex, DataColumnSubnetId, EthSpec, Slot};
+use types::{ChainSpec, ColumnIndex, DataColumnSubnetId, EthSpec, Slot};
 
 pub struct NetworkGlobals<E: EthSpec> {
     /// The current local ENR.
-    pub local_enr: RwLock<Enr>,
+    local_enr: RwLock<Enr>,
     /// The local peer_id.
     pub peer_id: RwLock<PeerId>,
     /// Listening multiaddrs.
     pub listen_multiaddrs: RwLock<Vec<Multiaddr>>,
     /// The collection of known peers.
     pub peers: RwLock<PeerDB<E>>,
-    // The local meta data of our node.
-    local_metadata: RwLock<MetaData<E>>,
     /// The current gossipsub topic subscriptions.
     pub gossipsub_subscriptions: RwLock<HashSet<GossipTopic>>,
     /// The current sync status of the node.
@@ -47,7 +46,6 @@ struct CustodyGroupCount {
 impl<E: EthSpec> NetworkGlobals<E> {
     pub fn new(
         enr: Enr,
-        local_metadata: MetaData<E>,
         trusted_peers: Vec<PeerId>,
         disable_peer_scoring: bool,
         config: Arc<NetworkConfig>,
@@ -78,7 +76,6 @@ impl<E: EthSpec> NetworkGlobals<E> {
             local_enr: RwLock::new(enr.clone()),
             peer_id: RwLock::new(enr.peer_id()),
             listen_multiaddrs: RwLock::new(Vec::new()),
-            local_metadata: RwLock::new(local_metadata),
             peers: RwLock::new(PeerDB::new(trusted_peers, disable_peer_scoring)),
             gossipsub_subscriptions: RwLock::new(HashSet::new()),
             sync_state: RwLock::new(SyncState::Stalled),
@@ -98,13 +95,38 @@ impl<E: EthSpec> NetworkGlobals<E> {
         self.local_enr.read().clone()
     }
 
+    pub fn set_enr(&self, enr: Enr) {
+        *self.local_enr.write() = enr;
+    }
+
     /// Returns the local libp2p PeerID.
     pub fn local_peer_id(&self) -> PeerId {
         *self.peer_id.read()
     }
 
     pub fn local_metadata(&self) -> MetaData<E> {
-        todo!();
+        let enr = self.local_enr();
+        let attnets = enr
+            .attestation_bitfield::<E>()
+            .unwrap_or(Default::default());
+        let syncnets = enr
+            .sync_committee_bitfield::<E>()
+            .unwrap_or(Default::default());
+
+        if self.spec.is_peer_das_scheduled() {
+            MetaData::V3(MetaDataV3 {
+                seq_number: enr.seq(),
+                attnets,
+                syncnets,
+                custody_group_count: self.public_custody_group_count(),
+            })
+        } else {
+            MetaData::V2(MetaDataV2 {
+                seq_number: enr.seq(),
+                attnets,
+                syncnets,
+            })
+        }
     }
 
     /// Returns the list of `Multiaddr` that the underlying libp2p instance is listening on.
@@ -129,6 +151,10 @@ impl<E: EthSpec> NetworkGlobals<E> {
         &self.all_sampling_columns[..self.all_sampling_columns.len().min(cgc)]
     }
 
+    fn public_custody_group_count(&self) -> u64 {
+        todo!();
+    }
+
     /// Returns the custody group count (CGC)
     fn custody_group_count(&self, slot: Slot) -> u64 {
         let cgc = self.custody_group_count.read().value;
@@ -143,20 +169,6 @@ impl<E: EthSpec> NetworkGlobals<E> {
             .expect("should compute node sampling size from valid chain spec")
     }
 
-    pub fn update_metadata_bitfields(
-        &self,
-        local_attnets: BitVector<E::SubnetBitfieldLength>,
-        local_syncnets: BitVector<E::SyncCommitteeSubnetCount>,
-    ) {
-        let mut meta_data_w = self.local_metadata.write();
-
-        *meta_data_w.seq_number_mut() += 1;
-        *meta_data_w.attnets_mut() = local_attnets;
-        if let Ok(syncnets) = meta_data_w.syncnets_mut() {
-            *syncnets = local_syncnets;
-        }
-    }
-
     /// Returns the number of libp2p connected peers.
     pub fn connected_peers(&self) -> usize {
         self.peers.read().connected_peer_ids().count()
@@ -245,7 +257,8 @@ impl<E: EthSpec> NetworkGlobals<E> {
 
     pub(crate) fn new_test_globals_with_metadata(
         trusted_peers: Vec<PeerId>,
-        metadata: MetaData<E>,
+        // TODO: todo! Apply to enr
+        _metadata: MetaData<E>,
         config: Arc<NetworkConfig>,
         spec: Arc<ChainSpec>,
     ) -> NetworkGlobals<E> {
@@ -253,7 +266,7 @@ impl<E: EthSpec> NetworkGlobals<E> {
         let keypair = libp2p::identity::secp256k1::Keypair::generate();
         let enr_key: discv5::enr::CombinedKey = discv5::enr::CombinedKey::from_secp256k1(&keypair);
         let enr = discv5::enr::Enr::builder().build(&enr_key).unwrap();
-        NetworkGlobals::new(enr, metadata, trusted_peers, false, config, spec)
+        NetworkGlobals::new(enr, trusted_peers, false, config, spec)
     }
 }