feat: operator doppelgänger protection with slot-based detection

anchor/client/src/lib.rs

@@ -587,6 +587,7 @@ impl Client {
             duties_service.clone(),
             database.watch(),
             is_synced.clone(),
+            doppelganger_service.clone(),
             executor.clone(),
             &spec,
         );

anchor/client/src/notifier.rs

@@ -2,6 +2,7 @@ use std::sync::Arc;
 
 use anchor_validator_store::AnchorValidatorStore;
 use database::NetworkState;
+use operator_doppelganger::OperatorDoppelgangerService;
 use slot_clock::SlotClock;
 use task_executor::TaskExecutor;
 use tokio::{
@@ -13,15 +14,65 @@ use types::{ChainSpec, EthSpec};
 
 use crate::duties_service::DutiesService;
 
-/// Spawns a notifier service which periodically logs information about the node. It reuses
-/// Lighthouse's `notifier_service` for most functionality but adds some Anchor-specific information
-/// and logic.
-///
-/// Executes [`notify`] once per slot, halfway into it.
+/// Represents whether the client is synced with the execution layer
+#[derive(Debug, Clone, Copy)]
+enum SyncState {
+    Syncing,
+    Synced,
+}
+
+impl SyncState {
+    fn from_bool(is_synced: bool) -> Self {
+        if is_synced {
+            Self::Synced
+        } else {
+            Self::Syncing
+        }
+    }
+}
+
+/// Represents the operator's presence on chain
+#[derive(Debug, Clone, Copy)]
+enum OperatorState {
+    NoOperator,
+    OperatorPresent { cluster_count: usize },
+}
+
+impl OperatorState {
+    fn from_option(operator_id: Option<ssv_types::OperatorId>, cluster_count: usize) -> Self {
+        if operator_id.is_some() {
+            Self::OperatorPresent { cluster_count }
+        } else {
+            Self::NoOperator
+        }
+    }
+}
+
+/// Represents the doppelgänger monitoring state
+#[derive(Debug, Clone, Copy)]
+enum DoppelgangerState {
+    NotMonitoring,
+    MonitoringForDoppelganger,
+}
+
+impl DoppelgangerState {
+    fn from_service(doppelganger_service: Option<&Arc<OperatorDoppelgangerService>>) -> Self {
+        if doppelganger_service
+            .map(|service| service.is_monitoring())
+            .unwrap_or(false)
+        {
+            Self::MonitoringForDoppelganger
+        } else {
+            Self::NotMonitoring
+        }
+    }
+}
+
 pub fn spawn_notifier<E: EthSpec, T: SlotClock + 'static>(
     duties_service: Arc<DutiesService<AnchorValidatorStore<T, E>, T>>,
     network_state: watch::Receiver<NetworkState>,
     synced: watch::Receiver<bool>,
+    doppelganger_service: Option<Arc<OperatorDoppelgangerService>>,
     executor: TaskExecutor,
     spec: &ChainSpec,
 ) {
@@ -32,7 +83,13 @@ pub fn spawn_notifier<E: EthSpec, T: SlotClock + 'static>(
             if let Some(duration_to_next_slot) = duties_service.slot_clock.duration_to_next_slot() {
                 // Sleep until the middle of the next slot
                 sleep(duration_to_next_slot + slot_duration / 2).await;
-                notify(&duties_service, &network_state, &synced).await;
+                notify(
+                    &duties_service,
+                    &network_state,
+                    &synced,
+                    doppelganger_service.as_ref(),
+                )
+                .await;
             } else {
                 error!("Failed to read slot clock");
                 // If we can't read the slot clock, just wait another slot.
@@ -45,16 +102,13 @@ pub fn spawn_notifier<E: EthSpec, T: SlotClock + 'static>(
     executor.spawn(interval_fut, "validator_notifier");
 }
 
-/// Performs a single notification routine.
-///
-/// This serves to notify the user of the current application status via `info` logging.
-/// Additionally, some metrics are recorded by the `validator_services`.
 async fn notify<E: EthSpec, T: SlotClock + 'static>(
     duties_service: &DutiesService<AnchorValidatorStore<T, E>, T>,
     network_state: &watch::Receiver<NetworkState>,
     synced: &watch::Receiver<bool>,
+    doppelganger_service: Option<&Arc<OperatorDoppelgangerService>>,
 ) {
-    // Scope needed as Rust complains about `state` being held across `await` if using `drop`
+    // Gather state information
     let (operator_id, cluster_count) = {
         let state = network_state.borrow();
         let operator_id = state.get_own_id();
@@ -71,16 +125,51 @@ async fn notify<E: EthSpec, T: SlotClock + 'static>(
 
     let is_synced = *synced.borrow();
 
-    match (operator_id, is_synced) {
-        (None, false) => info!("Syncing"),
-        (None, true) => info!("Synced, waiting for operator key to appear on chain"),
-        (Some(operator_id), false) => {
+    // Build layered state
+    let sync = SyncState::from_bool(is_synced);
+    let operator = OperatorState::from_option(operator_id, cluster_count);
+    let doppelganger = DoppelgangerState::from_service(doppelganger_service);
+
+    // Match on compositional state layers
+    match (&sync, &operator, &doppelganger, validator_count) {
+        (SyncState::Syncing, OperatorState::NoOperator, _, _) => {
+            info!("Syncing")
+        }
+        (SyncState::Syncing, OperatorState::OperatorPresent { .. }, _, _) => {
+            let operator_id = operator_id.expect("operator_id is Some in this branch");
             info!(%operator_id, "Operator present on chain, waiting for sync")
         }
-        (Some(operator_id), true) if validator_count > 0 => {
+        (SyncState::Synced, OperatorState::NoOperator, _, _) => {
+            info!("Synced, waiting for operator key to appear on chain")
+        }
+        (
+            SyncState::Synced,
+            OperatorState::OperatorPresent { cluster_count },
+            DoppelgangerState::MonitoringForDoppelganger,
+            count,
+        ) if count > 0 => {
+            let operator_id = operator_id.expect("operator_id is Some in this branch");
+            info!(
+                %operator_id,
+                cluster_count,
+                validator_count = count,
+                "Monitoring for operator doppelgänger (duties paused)"
+            )
+        }
+        (
+            SyncState::Synced,
+            OperatorState::OperatorPresent { cluster_count },
+            DoppelgangerState::NotMonitoring,
+            count,
+        ) if count > 0 => {
+            let operator_id = operator_id.expect("operator_id is Some in this branch");
             info!(%operator_id, cluster_count, "Operator active");
+            // Only call Lighthouse's notifier when we're actually performing duties
             validator_services::notifier_service::notify(duties_service).await;
         }
-        (Some(operator_id), true) => info!(%operator_id, "Operator ready, no validators assigned"),
+        (SyncState::Synced, OperatorState::OperatorPresent { .. }, _, _) => {
+            let operator_id = operator_id.expect("operator_id is Some in this branch");
+            info!(%operator_id, "Operator ready, no validators assigned")
+        }
     }
 }