Move notifier and latency service to validator_services (#7427)

We would like to reuse the `notifier` and `latency_service` in Anchor. To make this possible, this PR moves these from `validator_client` to `validator_services` and makes them use the new `ValidatorStore` trait is used so that the code can be reused in Anchor.

validator_client/src/lib.rs

@@ -1,7 +1,5 @@
 pub mod cli;
 pub mod config;
-mod latency;
-mod notifier;
 
 use crate::cli::ValidatorClient;
 pub use config::Config;
@@ -21,7 +19,6 @@ use environment::RuntimeContext;
 use eth2::{reqwest::ClientBuilder, BeaconNodeHttpClient, StatusCode, Timeouts};
 use initialized_validators::Error::UnableToOpenVotingKeystore;
 use lighthouse_validator_store::LighthouseValidatorStore;
-use notifier::spawn_notifier;
 use parking_lot::RwLock;
 use reqwest::Certificate;
 use slot_clock::SlotClock;
@@ -39,10 +36,12 @@ use tokio::{
 use tracing::{debug, error, info, warn};
 use types::{EthSpec, Hash256};
 use validator_http_api::ApiSecret;
+use validator_services::notifier_service::spawn_notifier;
 use validator_services::{
     attestation_service::{AttestationService, AttestationServiceBuilder},
     block_service::{BlockService, BlockServiceBuilder},
     duties_service::{self, DutiesService, DutiesServiceBuilder},
+    latency_service,
     preparation_service::{PreparationService, PreparationServiceBuilder},
     sync_committee_service::SyncCommitteeService,
 };
@@ -601,11 +600,17 @@ impl<E: EthSpec> ProductionValidatorClient<E> {
             info!("Doppelganger protection disabled.")
         }
 
-        spawn_notifier(self).map_err(|e| format!("Failed to start notifier: {}", e))?;
+        let context = self.context.service_context("notifier".into());
+        spawn_notifier(
+            self.duties_service.clone(),
+            context.executor,
+            &self.context.eth2_config.spec,
+        )
+        .map_err(|e| format!("Failed to start notifier: {}", e))?;
 
         if self.config.enable_latency_measurement_service {
-            latency::start_latency_service(
+            latency_service::start_latency_service(
-                self.context.clone(),
+                self.context.executor.clone(),
                 self.duties_service.slot_clock.clone(),
                 self.duties_service.beacon_nodes.clone(),
             );

validator_client/validator_services/Cargo.toml

@@ -14,11 +14,11 @@ graffiti_file = { workspace = true }
 logging = { workspace = true }
 parking_lot = { workspace = true }
 safe_arith = { workspace = true }
-slot_clock =  { workspace = true }
+slot_clock = { workspace = true }
 task_executor = { workspace = true }
-tokio =  { workspace = true }
+tokio = { workspace = true }
 tracing = { workspace = true }
-tree_hash =  { workspace = true }
+tree_hash = { workspace = true }
-types =  { workspace = true }
+types = { workspace = true }
 validator_metrics = { workspace = true }
 validator_store = { workspace = true }

validator_client/validator_services/src/latency_service.rs

@@ -1,10 +1,9 @@
 use beacon_node_fallback::BeaconNodeFallback;
-use environment::RuntimeContext;
 use slot_clock::SlotClock;
 use std::sync::Arc;
+use task_executor::TaskExecutor;
 use tokio::time::sleep;
 use tracing::debug;
-use types::EthSpec;
 
 /// The latency service will run 11/12ths of the way through the slot.
 pub const SLOT_DELAY_MULTIPLIER: u32 = 11;
@@ -12,8 +11,8 @@ pub const SLOT_DELAY_DENOMINATOR: u32 = 12;
 
 /// Starts a service that periodically checks the latency between the VC and the
 /// candidate BNs.
-pub fn start_latency_service<T: SlotClock + 'static, E: EthSpec>(
+pub fn start_latency_service<T: SlotClock + 'static>(
-    context: RuntimeContext<E>,
+    executor: TaskExecutor,
     slot_clock: T,
     beacon_nodes: Arc<BeaconNodeFallback<T>>,
 ) {
@@ -57,5 +56,5 @@ pub fn start_latency_service<T: SlotClock + 'static, E: EthSpec>(
         }
     };
 
-    context.executor.spawn(future, "latency");
+    executor.spawn(future, "latency");
 }

validator_client/validator_services/src/lib.rs

@@ -1,6 +1,8 @@
 pub mod attestation_service;
 pub mod block_service;
 pub mod duties_service;
+pub mod latency_service;
+pub mod notifier_service;
 pub mod preparation_service;
 pub mod sync;
 pub mod sync_committee_service;

validator_client/validator_services/src/notifier_service.rs

@@ -1,18 +1,20 @@
-use crate::{DutiesService, ProductionValidatorClient};
+use crate::duties_service::DutiesService;
-use lighthouse_validator_store::LighthouseValidatorStore;
-use metrics::set_gauge;
 use slot_clock::SlotClock;
+use std::sync::Arc;
+use task_executor::TaskExecutor;
 use tokio::time::{sleep, Duration};
 use tracing::{debug, error, info};
-use types::EthSpec;
+use types::{ChainSpec, EthSpec};
+use validator_metrics::set_gauge;
+use validator_store::ValidatorStore;
 
 /// Spawns a notifier service which periodically logs information about the node.
-pub fn spawn_notifier<E: EthSpec>(client: &ProductionValidatorClient<E>) -> Result<(), String> {
+pub fn spawn_notifier<S: ValidatorStore + 'static, T: SlotClock + 'static>(
-    let context = client.context.service_context("notifier".into());
+    duties_service: Arc<DutiesService<S, T>>,
-    let executor = context.executor.clone();
+    executor: TaskExecutor,
-    let duties_service = client.duties_service.clone();
+    spec: &ChainSpec,
-
+) -> Result<(), String> {
-    let slot_duration = Duration::from_secs(context.eth2_config.spec.seconds_per_slot);
+    let slot_duration = Duration::from_secs(spec.seconds_per_slot);
 
     let interval_fut = async move {
         loop {
@@ -33,9 +35,7 @@ pub fn spawn_notifier<E: EthSpec>(client: &ProductionValidatorClient<E>) -> Resu
 }
 
 /// Performs a single notification routine.
-async fn notify<T: SlotClock + 'static, E: EthSpec>(
+async fn notify<S: ValidatorStore, T: SlotClock + 'static>(duties_service: &DutiesService<S, T>) {
-    duties_service: &DutiesService<LighthouseValidatorStore<T, E>, T>,
-) {
     let (candidate_info, num_available, num_synced) =
         duties_service.beacon_nodes.get_notifier_info().await;
     let num_total = candidate_info.len();
@@ -102,7 +102,7 @@ async fn notify<T: SlotClock + 'static, E: EthSpec>(
     }
 
     if let Some(slot) = duties_service.slot_clock.now() {
-        let epoch = slot.epoch(E::slots_per_epoch());
+        let epoch = slot.epoch(S::E::slots_per_epoch());
 
         let total_validators = duties_service.total_validator_count();
         let proposing_validators = duties_service.proposer_count(epoch);