Support chain-supplied validator voluntary exits

Cargo.toml

@@ -25,6 +25,7 @@ members = [
     "anchor/signature_collector",
     "anchor/subnet_tracker",
     "anchor/validator_store",
+    "anchor/voluntary_exit",
 ]
 
 resolver = "2"
@@ -57,6 +58,7 @@ ssv_network_config = { path = "anchor/common/ssv_network_config" }
 ssv_types = { path = "anchor/common/ssv_types" }
 subnet_tracker = { path = "anchor/subnet_tracker" }
 version = { path = "anchor/common/version" }
+voluntary_exit = { path = "anchor/voluntary_exit" }
 
 beacon_node_fallback = { git = "https://github.com/sigp/lighthouse", rev = "2c153d7e" }
 bls = { git = "https://github.com/sigp/lighthouse", rev = "2c153d7e" }

anchor/client/Cargo.toml

@@ -50,4 +50,5 @@ unused_port = { workspace = true }
 validator_metrics = { workspace = true }
 validator_services = { workspace = true }
 version = { workspace = true }
+voluntary_exit = { workspace = true }
 zeroize = { workspace = true }

anchor/client/src/lib.rs

@@ -43,7 +43,7 @@ use task_executor::TaskExecutor;
 use tokio::{
     net::TcpListener,
     select,
-    sync::{mpsc, oneshot, oneshot::Receiver},
+    sync::{mpsc, mpsc::unbounded_channel, oneshot, oneshot::Receiver},
     time::sleep,
 };
 use tracing::{debug, error, info, warn};
@@ -57,6 +57,9 @@ use validator_services::{
     preparation_service::PreparationServiceBuilder,
     sync_committee_service::SyncCommitteeService,
 };
+use voluntary_exit::{
+    voluntary_exit_processor::start_exit_processor, voluntary_exit_tracker::VoluntaryExitTracker,
+};
 use zeroize::Zeroizing;
 
 /// The filename within the `validators` directory that contains the slashing protection DB.
@@ -350,11 +353,17 @@ impl Client {
         let index_sync_tx =
             start_validator_index_syncer(beacon_nodes.clone(), database.clone(), executor.clone());
 
+        // We create the channel here so that we can pass the receiver to the syncer. But we need to
+        // delay starting the voluntary exit processor until we have created the validator store.
+        let (exit_tx, exit_rx) = unbounded_channel();
+        let voluntary_exit_tracker = Arc::new(VoluntaryExitTracker::new());
+
         // Start syncer
         let (historic_finished_tx, historic_finished_rx) = oneshot::channel();
         let mut syncer = eth::SsvEventSyncer::new(
             database.clone(),
             index_sync_tx,
+            exit_tx,
             eth::Config {
                 http_urls: config.execution_nodes,
                 ws_url: config.execution_nodes_websocket,
@@ -482,6 +491,16 @@ impl Client {
             config.prefer_builder_proposals,
         );
 
+        start_exit_processor(
+            slot_clock.clone(),
+            E::slots_per_epoch(),
+            beacon_nodes.clone(),
+            validator_store.clone(),
+            exit_rx,
+            executor.clone(),
+            voluntary_exit_tracker.clone(),
+        );
+
         let selection_proof_config = SelectionProofConfig {
             lookahead_slot: 0,
             computation_offset: Duration::ZERO,

anchor/common/ssv_types/src/share.rs

@@ -6,7 +6,7 @@ use crate::{ClusterId, OperatorId};
 pub const ENCRYPTED_KEY_LENGTH: usize = 256;
 
 /// One of N shares of a split validator key.
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Eq, PartialEq, Hash)]
 pub struct Share {
     /// Public Key of the validator
     pub validator_pubkey: PublicKeyBytes,

anchor/database/src/lib.rs

@@ -41,10 +41,12 @@ const CONNECTION_TIMEOUT: Duration = Duration::from_secs(5);
 type Pool = r2d2::Pool<SqliteConnectionManager>;
 type PoolConn = r2d2::PooledConnection<SqliteConnectionManager>;
 
-/// All of the shares that belong to the current operator
-/// 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
+/// All the shares that belong to the current operator.
+/// IMPORTANT: There are parts of the code that assume this only contains shares that belong to the
+/// current operator. If this ever changes, make sure to update the code accordingly.
+/// Primary: public key of validator, uniquely identifies a share
+/// Secondary: cluster id, corresponds to a list of shares
+/// Tertiary: owner of the cluster, corresponds to a list of shares
 pub type ShareMultiIndexMap = MultiIndexMap<
     PublicKeyBytes,
     ClusterId,

anchor/database/src/state.rs

@@ -263,7 +263,7 @@ impl NetworkState {
             .map(|cluster| cluster.cluster_members)
     }
 
-    fn get_cluster_members_for_validator(
+    pub fn get_cluster_members_for_validator(
         &self,
         validator_pk: &PublicKeyBytes,
     ) -> Option<IndexSet<OperatorId>> {

anchor/eth/Cargo.toml

@@ -24,3 +24,4 @@ tokio = { workspace = true }
 tower = "0.5.2"
 tracing = { workspace = true }
 types = { workspace = true }
+voluntary_exit = { workspace = true }