refactor: Revamp startup sequence and check sync status (#372)

We want to be able to listen to the network without an Operator ID, for example to act as an exporter, or to already discover some peers during sync.

Also, we want to make sure to not do anything punishable (invalid network messages, signing slashable objects) if we are out of sync.


  - Add `OwnValidatorId` to allow managers and network to start eagerly
- Provide watcher for synced status, unifying `historic_finished_notify` and `operational_status`
- Avoid sending messages and starting duties for slashable objects if not synced
- Add sync status logging to notifier

Author: dknopik

Cargo.lock

@@ -117,6 +117,7 @@ indexmap = "2.7.0"
 libp2p = { version = "0.55", default-features = false }
 multiaddr = "0.18.2"
 num_cpus = "1"
+once_cell = "1.21.3"
 openssl = "0.10.72"
 parking_lot = "0.12"
 pbkdf2 = "0.12.2"

Cargo.toml

@@ -18,7 +18,7 @@ use beacon_node_fallback::{
 };
 pub use cli::Node;
 use config::Config;
-use database::NetworkDatabase;
+use database::{NetworkDatabase, OwnOperatorId};
 use duties_tracker::{duties_tracker::DutiesTracker, voluntary_exit_tracker::VoluntaryExitTracker};
 use eth::{
     index_sync::start_validator_index_syncer, voluntary_exit_processor::start_exit_processor,
@@ -38,17 +38,15 @@ use sensitive_url::SensitiveUrl;
 use signature_collector::SignatureCollectorManager;
 use slashing_protection::SlashingDatabase;
 use slot_clock::{SlotClock, SystemTimeSlotClock};
-use ssv_types::OperatorId;
 use subnet_service::{SUBNET_COUNT, SubnetId, start_subnet_service};
 use task_executor::TaskExecutor;
 use tokio::{
     net::TcpListener,
-    select,
-    sync::{mpsc, mpsc::unbounded_channel, oneshot, oneshot::Receiver},
+    sync::{mpsc, mpsc::unbounded_channel},
     time::sleep,
 };
 use tracing::{debug, error, info, warn};
-use types::{ChainSpec, EthSpec, Hash256};
+use types::{EthSpec, Hash256};
 use validator_metrics::set_gauge;
 use validator_services::{
     attestation_service::AttestationServiceBuilder,
@@ -381,7 +379,6 @@ impl Client {
         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,
@@ -390,15 +387,14 @@ impl Client {
                 http_urls: config.execution_nodes,
                 ws_url: config.execution_nodes_websocket,
                 network: config.global_config.ssv_network.clone(),
-                historic_finished_notify: Some(historic_finished_tx),
             },
         )
         .await
         .map_err(|e| format!("Unable to create syncer: {e}"))?;
 
-        // Access to the operational status of the sync. This can be passed around to condition
-        // duties based on the current status of the sync
-        let _operational_status = syncer.operational_status();
+        // Access to the sync status. This can be passed around to condition duties based on whether
+        // we are synced.
+        let is_synced = syncer.is_synced();
 
         executor.spawn(
             async move {
@@ -409,10 +405,7 @@ impl Client {
             "syncer",
         );
 
-        // Wait until we have an operator id and historical sync is done
-        let operator_id = wait_for_operator_id_and_sync(&database, historic_finished_rx, &spec)
-            .await
-            .ok_or("Failed waiting for operator id")?;
+        let operator_id = OwnOperatorId::new(database.watch());
 
         // Network sender/receiver
         let (network_tx, network_rx) = mpsc::channel::<(SubnetId, Vec<u8>)>(9001);
@@ -442,9 +435,10 @@ impl Client {
                 processor_senders.clone(),
                 network_tx.clone(),
                 key.clone(),
-                operator_id,
+                operator_id.clone(),
                 Some(message_validator.clone()),
                 SUBNET_COUNT,
+                is_synced.clone(),
             )?)
         } else {
             Arc::new(ImpostorMessageSender::new(network_tx.clone(), SUBNET_COUNT))
@@ -453,7 +447,7 @@ impl Client {
         // Create the signature collector
         let signature_collector = SignatureCollectorManager::new(
             processor_senders.clone(),
-            operator_id,
+            operator_id.clone(),
             config.global_config.ssv_network.ssv_domain_type.clone(),
             message_sender.clone(),
             slot_clock.clone(),
@@ -463,7 +457,7 @@ impl Client {
         // Create the qbft manager
         let qbft_manager = QbftManager::new(
             processor_senders.clone(),
-            operator_id,
+            operator_id.clone(),
             slot_clock.clone(),
             message_sender,
             config.global_config.ssv_network.ssv_domain_type.clone(),
@@ -528,6 +522,7 @@ impl Client {
             config.builder_proposals,
             config.builder_boost_factor,
             config.prefer_builder_proposals,
+            is_synced.clone(),
         );
 
         start_exit_processor(
@@ -566,6 +561,24 @@ impl Client {
             ctx.write().duties_service = Some(duties_service.clone());
         }
 
+        // Spawn notifier for logging and metrics
+        spawn_notifier(
+            duties_service.clone(),
+            database.watch(),
+            is_synced.clone(),
+            executor.clone(),
+            &spec,
+        );
+
+        // Wait for sync to complete before starting services
+        info!("Waiting for sync to complete before starting services...");
+        is_synced
+            .clone()
+            .wait_for(|&is_synced| is_synced)
+            .await
+            .map_err(|_| "Sync watch channel closed")?;
+        info!("Sync complete, starting services...");
+
         let mut block_service_builder = BlockServiceBuilder::new()
             .slot_clock(slot_clock.clone())
             .validator_store(validator_store.clone())
@@ -644,13 +657,6 @@ impl Client {
 
         http_api_shared_state.write().database_state = Some(database.watch());
 
-        spawn_notifier(
-            duties_service.clone(),
-            database.watch(),
-            executor.clone(),
-            &spec,
-        );
-
         if !config.disable_latency_measurement_service {
             start_latency_service(executor.clone(), slot_clock.clone(), beacon_nodes.clone());
         }
@@ -831,33 +837,6 @@ async fn poll_whilst_waiting_for_genesis(
     }
 }
 
-async fn wait_for_operator_id_and_sync(
-    database: &Arc<NetworkDatabase>,
-    mut sync_notification: Receiver<()>,
-    spec: &Arc<ChainSpec>,
-) -> Option<OperatorId> {
-    let sleep_duration = Duration::from_secs(spec.seconds_per_slot);
-    let mut state = database.watch();
-    let id = loop {
-        select! {
-            result = state.changed() => {
-                result.ok()?;
-                if let Some(id) = state.borrow().get_own_id() {
-                    break id;
-                }
-            }
-            _ = sleep(sleep_duration) => info!("Waiting for operator id"),
-        }
-    };
-    info!(id = *id, "Operator found on chain");
-    loop {
-        select! {
-            result = &mut sync_notification => return result.ok().map(|_| id),
-            _ = sleep(sleep_duration) => info!("Waiting for historical sync to finish"),
-        }
-    }
-}
-
 pub fn load_pem_certificate<P: AsRef<Path>>(pem_path: P) -> Result<Certificate, String> {
     let mut buf = Vec::new();
     File::open(&pem_path)

anchor/client/src/lib.rs

@@ -21,6 +21,7 @@ use crate::duties_service::DutiesService;
 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>,
     executor: TaskExecutor,
     spec: &ChainSpec,
 ) {
@@ -31,7 +32,7 @@ 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).await;
+                notify(&duties_service, &network_state, &synced).await;
             } else {
                 error!("Failed to read slot clock");
                 // If we can't read the slot clock, just wait another slot.
@@ -51,6 +52,7 @@ pub fn spawn_notifier<E: EthSpec, T: SlotClock + 'static>(
 async fn notify<E: EthSpec, T: SlotClock + 'static>(
     duties_service: &DutiesService<AnchorValidatorStore<T, E>, T>,
     network_state: &watch::Receiver<NetworkState>,
+    synced: &watch::Receiver<bool>,
 ) {
     // Scope needed as Rust complains about `state` being held across `await` if using `drop`
     let (operator_id, cluster_count) = {
@@ -60,14 +62,18 @@ async fn notify<E: EthSpec, T: SlotClock + 'static>(
         (operator_id, cluster_count)
     };
 
-    if let Some(operator_id) = operator_id {
-        if duties_service.total_validator_count() > 0 {
+    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) => {
+            info!(%operator_id, "Operator present on chain, waiting for sync")
+        }
+        (Some(operator_id), true) if duties_service.total_validator_count() > 0 => {
             info!(%operator_id, cluster_count, "Operator active");
             validator_services::notifier_service::notify(duties_service).await;
-        } else {
-            info!(%operator_id, "Operator ready, no validators assigned");
         }
-    } else {
-        error!("No operator ID - the operator might have been removed via the contract");
+        (Some(operator_id), true) => info!(%operator_id, "Operator ready, no validators assigned"),
     }
 }

anchor/client/src/notifier.rs

@@ -6,6 +6,7 @@ authors = ["Sigma Prime <[email protected]>"]
 
 [dependencies]
 base64 = { workspace = true }
+once_cell = { workspace = true }
 openssl = { workspace = true }
 r2d2 = { workspace = true }
 r2d2_sqlite = { workspace = true }

anchor/database/Cargo.toml

@@ -5,6 +5,7 @@ use std::{
     time::Duration,
 };
 
+use once_cell::sync::OnceCell;
 use openssl::{pkey::Public, rsa::Rsa};
 use r2d2_sqlite::SqliteConnectionManager;
 use rusqlite::{Transaction, params};
@@ -228,3 +229,54 @@ impl NetworkDatabase {
         });
     }
 }
+
+/// A helper to get the operator ID of the current operator. Caches the ID after successfully
+/// retrieving it to avoid locking the state further.
+#[derive(Clone)]
+pub enum OwnOperatorId {
+    /// The operator ID was known when the `OwnOperatorId` was created.
+    Known(OperatorId),
+    /// The operator ID was not known when the `OwnOperatorId` was created. It will be retrieved
+    /// from the `receiver` and cached in the `id` on first success.
+    FromState {
+        receiver: Receiver<NetworkState>,
+        /// We use a `OnceLock` so that `get` can be called without a mutable reference.
+        id: OnceCell<OperatorId>,
+    },
+}
+
+impl OwnOperatorId {
+    /// Creates the `OwnOperatorId` to either immediately store the operator ID or to recheck it on
+    /// later `get` calls.
+    pub fn new(receiver: Receiver<NetworkState>) -> Self {
+        if let Some(operator_id) = receiver.borrow().get_own_id() {
+            Self::Known(operator_id)
+        } else {
+            Self::FromState {
+                receiver,
+                id: OnceCell::new(),
+            }
+        }
+    }
+
+    /// Get the operator ID if it is available. Caches the ID internally after the first successful
+    /// call to avoid locking the state in the future. This is possible because the own Operator ID
+    /// never changes.
+    pub fn get(&self) -> Option<OperatorId> {
+        match self {
+            Self::Known(id) => Some(*id),
+            Self::FromState { receiver, id } => {
+                // Switch to `std`'s OnceLock as soon as `get_or_try_init` is stable
+                id.get_or_try_init(|| receiver.borrow().get_own_id().ok_or(()))
+                    .ok()
+                    .copied()
+            }
+        }
+    }
+}
+
+impl From<OperatorId> for OwnOperatorId {
+    fn from(operator_id: OperatorId) -> Self {
+        Self::Known(operator_id)
+    }
+}