Enable notifier and latency service

anchor/client/src/cli.rs

@@ -525,6 +525,14 @@ pub struct Node {
     )]
     pub prefer_builder_proposals: bool,
 
+    #[clap(
+        long,
+        help = "Disable the latency measurement service.",
+        display_order = 0,
+        help_heading = FLAG_HEADER
+    )]
+    pub disable_latency_measurement_service: bool,
+
     #[clap(flatten)]
     pub logging_flags: LoggingFlags,
 }

anchor/client/src/config.rs

@@ -73,6 +73,8 @@ pub struct Config {
     pub builder_boost_factor: Option<u64>,
     /// Should external payloads always be preferred
     pub prefer_builder_proposals: bool,
+    /// Controls whether the latency measurement service is enabled
+    pub disable_latency_measurement_service: bool,
 }
 
 impl Config {
@@ -126,6 +128,7 @@ impl Config {
             builder_boost_factor: None,
             prefer_builder_proposals: false,
             gas_limit: 36_000_000,
+            disable_latency_measurement_service: false,
         }
     }
 }
@@ -262,8 +265,8 @@ pub fn from_cli(cli_args: &Node) -> Result<Config, String> {
 
     config.enable_high_validator_count_metrics = cli_args.enable_high_validator_count_metrics;
 
-    // debugging stuff
     config.impostor = cli_args.impostor.map(OperatorId);
+    config.disable_latency_measurement_service = cli_args.disable_latency_measurement_service;
 
     // Performance options
     if let Some(max_workers) = cli_args.max_workers {

anchor/client/src/lib.rs

@@ -57,6 +57,8 @@ use validator_services::{
     block_service::BlockServiceBuilder,
     duties_service,
     duties_service::{DutiesServiceBuilder, SelectionProofConfig},
+    latency_service::start_latency_service,
+    notifier_service::spawn_notifier,
     preparation_service::PreparationServiceBuilder,
     sync_committee_service::SyncCommitteeService,
 };
@@ -605,19 +607,13 @@ impl Client {
             .map_err(|e| format!("Unable to start preparation service: {e}"))?;
 
         http_api_shared_state.write().database_state = Some(database.watch());
-        // TODO: reuse this from lighthouse
-        // https://github.com/sigp/anchor/issues/251
-        // spawn_notifier(self).map_err(|e| format!("Failed to start notifier: {e}"))?;
-
-        // TODO: reuse this from lighthouse
-        // https://github.com/sigp/anchor/issues/250
-        // if self.config.enable_latency_measurement_service {
-        //     latency::start_latency_service(
-        //         self.context.clone(),
-        //         self.duties_service.slot_clock.clone(),
-        //         self.duties_service.beacon_nodes.clone(),
-        //     );
-        // }
+
+        spawn_notifier(duties_service.clone(), executor.clone(), &spec)
+            .map_err(|e| format!("Failed to start notifier: {e}"))?;
+
+        if !config.disable_latency_measurement_service {
+            start_latency_service(executor.clone(), slot_clock.clone(), beacon_nodes.clone());
+        }
 
         Ok(())
     }

anchor/validator_store/src/lib.rs

@@ -733,13 +733,16 @@ impl<T: SlotClock, E: EthSpec> ValidatorStore for AnchorValidatorStore<T, E> {
             .and_then(|v| v.metadata.index.map(|idx| *idx as u64))
     }
 
-    fn voting_pubkeys<I, F>(&self, _filter_func: F) -> I
+    fn voting_pubkeys<I, F>(&self, filter_func: F) -> I
     where
         I: FromIterator<PublicKeyBytes>,
         F: Fn(DoppelgangerStatus) -> Option<PublicKeyBytes>,
     {
-        // we don't care about doppelgangers
-        self.validators.iter().map(|v| *v.key()).collect()
+        // Treat all validators as `SigningEnabled`
+        self.validators
+            .iter()
+            .filter_map(|v| filter_func(DoppelgangerStatus::SigningEnabled(*v.key())))
+            .collect()
     }
 
     fn doppelganger_protection_allows_signing(&self, _validator_pubkey: PublicKeyBytes) -> bool {