Set topic parameters

Cargo.toml

@@ -24,7 +24,7 @@ members = [
     "anchor/processor",
     "anchor/qbft_manager",
     "anchor/signature_collector",
-    "anchor/subnet_tracker",
+    "anchor/subnet_service",
     "anchor/validator_store",
 ]
 
@@ -58,7 +58,7 @@ qbft_manager = { path = "anchor/qbft_manager" }
 signature_collector = { path = "anchor/signature_collector" }
 ssv_network_config = { path = "anchor/common/ssv_network_config" }
 ssv_types = { path = "anchor/common/ssv_types" }
-subnet_tracker = { path = "anchor/subnet_tracker" }
+subnet_service = { path = "anchor/subnet_service" }
 version = { path = "anchor/common/version" }
 
 beacon_node_fallback = { git = "https://github.com/sigp/lighthouse", rev = "9b84dac" }

anchor/client/Cargo.toml

@@ -43,7 +43,7 @@ slot_clock = { workspace = true }
 ssv_network_config = { workspace = true }
 ssv_types = { workspace = true }
 strum = { workspace = true }
-subnet_tracker = { workspace = true }
+subnet_service = { workspace = true }
 task_executor = { workspace = true }
 tokio = { workspace = true }
 tracing = { workspace = true }

anchor/client/src/cli.rs

@@ -558,11 +558,20 @@ pub struct Node {
 
     #[clap(
         long,
-        help = "Disables peer scoring altogether.",
+        help = "Disables gossipsub peer scoring.",
         display_order = 0,
         help_heading = FLAG_HEADER
     )]
-    pub disable_peer_scoring: bool,
+    pub disable_gossipsub_peer_scoring: bool,
+
+    #[clap(
+        long,
+        help = "Disables gossipsub topic scoring.",
+        action = ArgAction::Set,
+        default_value = "true",
+        hide = true
+    )]
+    pub disable_gossipsub_topic_scoring: bool,
 
     #[clap(flatten)]
     pub logging_flags: LoggingFlags,

anchor/client/src/config.rs

@@ -216,7 +216,8 @@ pub fn from_cli(cli_args: &Node) -> Result<Config, String> {
     config.network.subscribe_all_subnets = cli_args.subscribe_all_subnets;
 
     // Network related - set peer scoring configuration
-    config.network.disable_peer_scoring = cli_args.disable_peer_scoring;
+    config.network.disable_gossipsub_peer_scoring = cli_args.disable_gossipsub_peer_scoring;
+    config.network.disable_gossipsub_topic_scoring = cli_args.disable_gossipsub_topic_scoring;
 
     config.beacon_nodes_tls_certs = cli_args.beacon_nodes_tls_certs.clone();
     config.execution_nodes_tls_certs = cli_args.execution_nodes_tls_certs.clone();

anchor/client/src/lib.rs

@@ -40,7 +40,7 @@ use signature_collector::SignatureCollectorManager;
 use slashing_protection::SlashingDatabase;
 use slot_clock::{SlotClock, SystemTimeSlotClock};
 use ssv_types::OperatorId;
-use subnet_tracker::{SubnetId, start_subnet_tracker};
+use subnet_service::{SUBNET_COUNT, SubnetId, start_subnet_service};
 use task_executor::TaskExecutor;
 use tokio::{
     net::TcpListener,
@@ -200,13 +200,6 @@ impl Client {
             .map_err(|e| format!("Unable to open Anchor database: {e}"))?,
         );
 
-        let subnet_tracker = start_subnet_tracker(
-            database.watch(),
-            network::SUBNET_COUNT,
-            config.network.subscribe_all_subnets,
-            &executor,
-        );
-
         // Initialize slashing protection.
         let slashing_db_path = config.data_dir.join(SLASHING_PROTECTION_FILENAME);
         let slashing_protection =
@@ -436,13 +429,10 @@ impl Client {
                 key.clone(),
                 operator_id,
                 Some(message_validator.clone()),
-                network::SUBNET_COUNT,
+                SUBNET_COUNT,
             )?)
         } else {
-            Arc::new(ImpostorMessageSender::new(
-                network_tx.clone(),
-                network::SUBNET_COUNT,
-            ))
+            Arc::new(ImpostorMessageSender::new(network_tx.clone(), SUBNET_COUNT))
         };
 
         // Create the signature collector
@@ -465,6 +455,17 @@ impl Client {
         )
         .map_err(|e| format!("Unable to initialize qbft manager: {e:?}"))?;
 
+        // Start the subnet service now that we have slot_clock
+        let subnet_service = start_subnet_service::<E>(
+            database.watch(),
+            SUBNET_COUNT,
+            config.network.subscribe_all_subnets,
+            config.network.disable_gossipsub_topic_scoring,
+            &executor,
+            slot_clock.clone(),
+            spec.clone(),
+        );
+
         let (outcome_tx, outcome_rx) = mpsc::channel::<message_receiver::Outcome>(9000);
 
         let message_receiver = NetworkMessageReceiver::new(
@@ -479,12 +480,12 @@ impl Client {
         // Start the p2p network
         let mut network = Network::try_new::<E>(
             &config.network,
-            subnet_tracker,
+            subnet_service,
             network_rx,
             Arc::new(message_receiver),
             outcome_rx,
             executor.clone(),
-            &spec,
+            spec.clone(),
         )
         .await
         .map_err(|e| format!("Unable to start network: {e}"))?;
@@ -495,7 +496,7 @@ impl Client {
         }
 
         // Spawn the network listening task
-        executor.spawn(network.run(), "network");
+        executor.spawn(network.run::<E>(), "network");
 
         let validator_store = AnchorValidatorStore::<_, E>::new(
             database.watch(),

anchor/message_sender/Cargo.toml

@@ -14,6 +14,6 @@ openssl = { workspace = true }
 processor = { workspace = true }
 slot_clock = { workspace = true }
 ssv_types = { workspace = true }
-subnet_tracker = { workspace = true }
+subnet_service = { workspace = true }
 tokio = { workspace = true }
 tracing = { workspace = true }

anchor/message_sender/src/impostor.rs

@@ -1,5 +1,5 @@
 use ssv_types::{CommitteeId, consensus::UnsignedSSVMessage, message::SignedSSVMessage};
-use subnet_tracker::SubnetId;
+use subnet_service::SubnetId;
 use tokio::sync::mpsc;
 use tracing::debug;
 

anchor/message_sender/src/network.rs

@@ -13,7 +13,7 @@ use ssv_types::{
     CommitteeId, OperatorId, consensus::UnsignedSSVMessage, message::SignedSSVMessage,
 };
 use ssz::Encode;
-use subnet_tracker::SubnetId;
+use subnet_service::SubnetId;
 use tokio::sync::{mpsc, mpsc::error::TrySendError};
 use tracing::{debug, error, warn};
 

anchor/network/Cargo.toml

@@ -33,7 +33,7 @@ serde = { workspace = true }
 serde_json = "1.0.137"
 ssv_types = { workspace = true }
 ssz_types = "0.10"
-subnet_tracker = { workspace = true }
+subnet_service = { workspace = true }
 task_executor = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }

anchor/network/src/behaviour.rs

@@ -102,7 +102,7 @@ impl AnchorBehaviour {
         .map_err(|e| Gossipsub(e.to_string()))?;
 
         // Add peer scoring if not disabled
-        if !network_config.disable_peer_scoring {
+        if !network_config.disable_gossipsub_peer_scoring {
             let slots_per_epoch = E::slots_per_epoch();
             let slot_duration = Duration::from_secs(spec.seconds_per_slot);
             let one_epoch_duration = slot_duration * slots_per_epoch as u32;

anchor/network/src/config.rs

@@ -54,8 +54,11 @@ pub struct Config {
     /// List of nodes to initially connect to, on Multiaddr format.
     pub boot_nodes_multiaddr: Vec<Multiaddr>,
 
-    /// Disables peer scoring altogether.
-    pub disable_peer_scoring: bool,
+    /// Disables gossipsub peer scoring altogether.
+    pub disable_gossipsub_peer_scoring: bool,
+
+    /// Disables gossipsub topic scoring and message rate calculations.
+    pub disable_gossipsub_topic_scoring: bool,
 
     /// Disables the discovery protocol from starting.
     pub disable_discovery: bool,
@@ -103,7 +106,8 @@ impl Default for Config {
             target_peers: 50,
             boot_nodes_enr: vec![],
             boot_nodes_multiaddr: vec![],
-            disable_peer_scoring: false,
+            disable_gossipsub_peer_scoring: false,
+            disable_gossipsub_topic_scoring: true,
             disable_discovery: false,
             disable_quic_support: false,
             subscribe_all_subnets: false,

anchor/network/src/discovery.rs

@@ -37,7 +37,7 @@ use lighthouse_network::{
 use ssv_types::domain_type::DomainType;
 use ssz::{Decode, Encode};
 use ssz_types::{BitVector, Bitfield, length::Fixed, typenum::U128};
-use subnet_tracker::SubnetId;
+use subnet_service::SubnetId;
 use tokio::sync::mpsc;
 use tracing::{debug, error, info, trace, warn};
 

anchor/network/src/handshake/node_info.rs

@@ -1,15 +1,12 @@
 use discv5::libp2p_identity::{Keypair, SigningError};
 use serde::{Deserialize, Serialize};
 use serde_json;
-use subnet_tracker::SubnetId;
+use subnet_service::{SubnetBits, SubnetId};
 use thiserror::Error;
 
-use crate::{
-    SubnetBits,
-    handshake::{
-        envelope::{Envelope, make_unsigned},
-        node_info::Error::Validation,
-    },
+use crate::handshake::{
+    envelope::{Envelope, make_unsigned},
+    node_info::Error::Validation,
 };
 
 #[derive(Debug, Error)]

anchor/network/src/lib.rs

@@ -14,6 +14,3 @@ pub use config::Config;
 pub use lighthouse_network::{ListenAddr, ListenAddress};
 pub use network::Network;
 pub type Enr = discv5::enr::Enr<discv5::enr::CombinedKey>;
-
-pub const SUBNET_COUNT: usize = 128;
-type SubnetBits = [u8; SUBNET_COUNT / 8];