Introduce validator index syncer (sigp#212)

Co-authored-by: diegomrsantos <diegomrsantos@gmail.com>

Author: dknopik

Cargo.lock

@@ -109,11 +109,20 @@ pub struct Node {
     #[clap(
         long,
         value_name = "NETWORK_ADDRESSES",
-        help = "Comma-separated addresses to one or more beacon node HTTP APIs. \
+        help = "Comma-separated addresses to one or more execution node JSON-RPC APIs. \
                 Default is http://localhost:8545.",
         display_order = 0
     )]
-    pub execution_nodes: Option<Vec<String>>,
+    pub execution_rpc: Option<Vec<String>>,
+
+    #[clap(
+        long,
+        value_name = "NETWORK_ADDRESSES",
+        help = "Comma-separated addresses to one or more execution node WS APIs. \
+                Default is ws://localhost:8546.",
+        display_order = 0
+    )]
+    pub execution_ws: Option<Vec<String>>,
 
     #[clap(
         long,

anchor/client/src/cli.rs

@@ -36,6 +36,8 @@ pub struct Config {
     pub proposer_nodes: Vec<SensitiveUrl>,
     /// The http endpoints of the execution node APIs.
     pub execution_nodes: Vec<SensitiveUrl>,
+    /// The websocket endpoints of the execution node APIs.
+    pub execution_nodes_websocket: Vec<SensitiveUrl>,
     /// beacon node is not synced at startup.
     pub allow_unsynced_beacon_node: bool,
     /// If true, use longer timeouts for requests made to the beacon node.
@@ -79,19 +81,18 @@ impl Config {
 
         let beacon_nodes = vec![SensitiveUrl::parse(DEFAULT_BEACON_NODE)
             .expect("beacon_nodes must always be a valid url.")];
-        let execution_nodes = vec![
-            SensitiveUrl::parse(DEFAULT_EXECUTION_NODE)
-                .expect("execution_nodes must always be a valid url."),
-            SensitiveUrl::parse(DEFAULT_EXECUTION_NODE_WS)
-                .expect("execution_nodes must always be a valid url."),
-        ];
+        let execution_nodes = vec![SensitiveUrl::parse(DEFAULT_EXECUTION_NODE)
+            .expect("execution_nodes must always be a valid url.")];
+        let execution_nodes_websocket = vec![SensitiveUrl::parse(DEFAULT_EXECUTION_NODE_WS)
+            .expect("execution_nodes_websocket must always be a valid url.")];
 
         Self {
             data_dir,
             ssv_network,
             beacon_nodes,
             proposer_nodes: vec![],
             execution_nodes,
+            execution_nodes_websocket,
             allow_unsynced_beacon_node: false,
             use_long_timeouts: false,
             http_api: <_>::default(),
@@ -127,20 +128,18 @@ pub fn from_cli(cli_args: &Node) -> Result<Config, String> {
             .map_err(|e| format!("Failed to create {:?}: {:?}", config.data_dir, e))?;
     }
 
-    if let Some(beacon_nodes) = &cli_args.beacon_nodes {
-        config.beacon_nodes = beacon_nodes
-            .iter()
-            .map(|s| SensitiveUrl::parse(s))
-            .collect::<Result<_, _>>()
-            .map_err(|e| format!("Unable to parse beacon node URL: {:?}", e))?;
+    if let Some(ref beacon_nodes) = cli_args.beacon_nodes {
+        parse_urls(&mut config.beacon_nodes, beacon_nodes, "beacon node")?;
     }
-
-    if let Some(execution_nodes) = &cli_args.execution_nodes {
-        config.execution_nodes = execution_nodes
-            .iter()
-            .map(|s| SensitiveUrl::parse(s))
-            .collect::<Result<_, _>>()
-            .map_err(|e| format!("Unable to parse execution node URL: {:?}", e))?;
+    if let Some(ref execution_rpc) = cli_args.execution_rpc {
+        parse_urls(&mut config.execution_nodes, execution_rpc, "execution RPC")?;
+    }
+    if let Some(ref execution_ws) = cli_args.execution_ws {
+        parse_urls(
+            &mut config.execution_nodes_websocket,
+            execution_ws,
+            "execution WebSocket",
+        )?;
     }
 
     // Password to decrypt rsa key file
@@ -236,6 +235,16 @@ pub fn from_cli(cli_args: &Node) -> Result<Config, String> {
     Ok(config)
 }
 
+/// Read SensitiveUrls from given CLI Strings
+fn parse_urls(dest: &mut Vec<SensitiveUrl>, src: &[String], kind: &str) -> Result<(), String> {
+    *dest = src
+        .iter()
+        .map(|s| SensitiveUrl::parse(s))
+        .collect::<Result<_, _>>()
+        .map_err(|e| format!("Unable to parse {kind} URL: {:?}", e))?;
+    Ok(())
+}
+
 /// Gets the listening_addresses for lighthouse based on the cli options.
 pub fn parse_listening_addresses(cli_args: &Node) -> Result<ListenAddress, String> {
     // parse the possible ips

anchor/client/src/config.rs

@@ -20,6 +20,7 @@ use beacon_node_fallback::{
 pub use cli::Node;
 use config::Config;
 use database::NetworkDatabase;
+use eth::index_sync::start_validator_index_syncer;
 use eth2::{
     reqwest::{Certificate, ClientBuilder},
     BeaconNodeHttpClient, Timeouts,
@@ -105,6 +106,7 @@ impl Client {
         info!(
             beacon_nodes = format!("{:?}", &config.beacon_nodes),
             execution_nodes = format!("{:?}", &config.execution_nodes),
+            execution_nodes_websocket = format!("{:?}", &config.execution_nodes_websocket),
             data_dir = format!("{:?}", config.data_dir),
             "Starting the Anchor client"
         );
@@ -322,25 +324,26 @@ impl Client {
         // Wait until genesis has occurred.
         wait_for_genesis(&beacon_nodes, genesis_time).await?;
 
+        // Start validator index syncer
+        let index_sync_tx =
+            start_validator_index_syncer(beacon_nodes.clone(), database.clone(), executor.clone());
+
         // Start syncer
         let (historic_finished_tx, historic_finished_rx) = oneshot::channel();
         let mut syncer = eth::SsvEventSyncer::new(
             database.clone(),
-            // TODO this is very hacky, but `eth::Config` has a TODO anyways
+            index_sync_tx,
             eth::Config {
                 http_url: config
                     .execution_nodes
                     .first()
                     .ok_or("No execution node http url specified")?
-                    .full
-                    .to_string(),
+                    .clone(),
                 ws_url: config
-                    .execution_nodes
-                    .get(1)
-                    .ok_or("No execution node wss url specified")?
-                    .full
-                    .to_string(),
-                beacon_url: "".to_string(), // this one is not actually needed :)
+                    .execution_nodes_websocket
+                    .first()
+                    .ok_or("No execution node ws url specified")?
+                    .clone(),
                 network: config.ssv_network.clone(),
                 historic_finished_notify: Some(historic_finished_tx),
             },

anchor/client/src/lib.rs

@@ -78,7 +78,7 @@ pub struct ValidatorMetadata {
     /// The cluster that is responsible for this validator
     pub cluster_id: ClusterId,
     /// Index of the validator
-    pub index: ValidatorIndex,
+    pub index: Option<ValidatorIndex>,
     /// Graffiti
     pub graffiti: Graffiti,
 }

anchor/common/ssv_types/src/cluster.rs

@@ -111,7 +111,7 @@ impl TryFrom<&Row<'_>> for ValidatorMetadata {
         let cluster_id: ClusterId = ClusterId(row.get(1)?);
 
         // Get ValidatorIndex from column 2
-        let index: ValidatorIndex = ValidatorIndex(row.get(2)?);
+        let index = row.get::<_, Option<usize>>(2)?.map(ValidatorIndex);
 
         // Get Graffiti from column 3
         let graffiti = Graffiti(row.get::<_, [u8; GRAFFITI_BYTES_LEN]>(3)?);

anchor/common/ssv_types/src/sql_conversions.rs

@@ -13,6 +13,7 @@ rand = { workspace = true }
 rusqlite = { workspace = true }
 ssv_types = { workspace = true }
 tokio = { workspace = true }
+tracing = { workspace = true }
 types = { workspace = true }
 
 [dev-dependencies]

anchor/database/Cargo.toml

@@ -29,7 +29,7 @@ impl NetworkDatabase {
             .execute(params![
                 validator.public_key.to_string(), // validator public key
                 *cluster.cluster_id,              // cluster id
-                *validator.index,                 // validator index
+                validator.index.as_deref(),       // validator index
                 validator.graffiti.0.as_slice(),  // graffiti
             ])?;
 

anchor/database/src/cluster_operations.rs

@@ -45,7 +45,7 @@ type PoolConn = r2d2::PooledConnection<SqliteConnectionManager>;
 /// Primary: public key of validator. uniquely identifies share
 /// Secondary: cluster id. corresponds to a list of shares
 /// Tertiary: owner of the cluster. corresponds to a list of shares
-pub(crate) type ShareMultiIndexMap = MultiIndexMap<
+pub type ShareMultiIndexMap = MultiIndexMap<
     PublicKeyBytes,
     ClusterId,
     Address,
@@ -59,7 +59,7 @@ pub(crate) type ShareMultiIndexMap = MultiIndexMap<
 /// Primary: public key of the validator. uniquely identifies the metadata
 /// Secondary: cluster id. corresponds to list of metadata for all validators
 /// Tertiary: owner of the cluster: corresponds to list of metadata for all validators
-pub(crate) type MetadataMultiIndexMap = MultiIndexMap<
+pub type MetadataMultiIndexMap = MultiIndexMap<
     PublicKeyBytes,
     ClusterId,
     Address,
@@ -73,7 +73,7 @@ pub(crate) type MetadataMultiIndexMap = MultiIndexMap<
 /// Primary: cluster id. uniquely identifies a cluster
 /// Secondary: public key of the validator. uniquely identifies a cluster
 /// Tertiary: owner of the cluster. uniquely identifies a cluster
-pub(crate) type ClusterMultiIndexMap = MultiIndexMap<
+pub type ClusterMultiIndexMap = MultiIndexMap<
     ClusterId,
     PublicKeyBytes,
     Address,

anchor/database/src/lib.rs

@@ -236,6 +236,10 @@ where
         // Only update the value in primary storage
         self.maps.primary.insert(k1.clone(), new_value)
     }
+
+    pub fn values(&self) -> impl Iterator<Item = &V> {
+        self.maps.primary.values()
+    }
 }
 
 // Implement unique access for primary key.