implement using state watcher

Cargo.toml

@@ -14,6 +14,7 @@ members = [
     "anchor/processor",
     "anchor/qbft_manager",
     "anchor/signature_collector",
+    "anchor/subnet_tracker",
     "anchor/validator_store",
 ]
 resolver = "2"
@@ -34,6 +35,7 @@ processor = { path = "anchor/processor" }
 qbft = { path = "anchor/common/qbft" }
 qbft_manager = { path = "anchor/qbft_manager" }
 signature_collector = { path = "anchor/signature_collector" }
+subnet_tracker = { path = "anchor/subnet_tracker" }
 ssv_types = { path = "anchor/common/ssv_types" }
 version = { path = "anchor/common/version" }
 

anchor/client/Cargo.toml

@@ -35,6 +35,7 @@ slashing_protection = { workspace = true }
 slot_clock = { workspace = true }
 ssv_types = { workspace = true }
 strum = { workspace = true }
+subnet_tracker = { workspace = true }
 task_executor = { workspace = true }
 tokio = { workspace = true }
 tracing = { workspace = true }

anchor/client/src/lib.rs

@@ -29,6 +29,7 @@ use std::net::SocketAddr;
 use std::path::Path;
 use std::sync::Arc;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
+use subnet_tracker::start_subnet_tracker;
 use task_executor::TaskExecutor;
 use tokio::net::TcpListener;
 use tokio::select;
@@ -135,17 +136,20 @@ impl Client {
             return Err("HTTP API Failed".to_string());
         }
 
-        // Start the p2p network
-        let network = Network::try_new(&config.network, executor.clone()).await?;
-        // Spawn the network listening task
-        executor.spawn(network.run(), "network");
-
         // Open database
         let database = Arc::new(
             NetworkDatabase::new(config.data_dir.join("anchor_db.sqlite").as_path(), &pubkey)
                 .map_err(|e| format!("Unable to open Anchor database: {e}"))?,
         );
 
+        let subnet_tracker =
+            start_subnet_tracker(database.watch(), network::SUBNET_COUNT, &executor);
+
+        // Start the p2p network
+        let network = Network::try_new(&config.network, subnet_tracker, executor.clone()).await?;
+        // Spawn the network listening task
+        executor.spawn(network.run(), "network");
+
         // Initialize slashing protection.
         let slashing_db_path = config.data_dir.join(SLASHING_PROTECTION_FILENAME);
         let slashing_protection =

anchor/network/Cargo.toml

@@ -7,7 +7,6 @@ authors = ["Sigma Prime <contact@sigmaprime.io>"]
 [dependencies]
 dirs = { workspace = true }
 discv5 = { workspace = true }
-ethereum_serde_utils = "1.0.0-beta.0"
 ethereum_ssz = "0.8.1"
 ethereum_ssz_derive = "0.8.1"
 futures = { workspace = true }
@@ -27,6 +26,7 @@ libp2p = { version = "0.54", default-features = false, features = [
 lighthouse_network = { workspace = true }
 serde = { workspace = true }
 ssz_types = "0.10"
+subnet_tracker = { workspace = true }
 task_executor = { workspace = true }
 tokio = { workspace = true }
 tracing = { workspace = true }