Merge remote-tracking branch 'origin/unstable' into dvt

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 }
-types =  { workspace = true }
+tree_hash = { workspace = true }
+types = { workspace = true }
 validator_metrics = { workspace = true }
 validator_store = { workspace = true }

validator_client/validator_services/src/attestation_service.rs

@@ -1,6 +1,5 @@
 use crate::duties_service::{DutiesService, DutyAndProof};
 use beacon_node_fallback::{ApiTopic, BeaconNodeFallback};
-use either::Either;
 use futures::future::join_all;
 use logging::crit;
 use slot_clock::SlotClock;
@@ -461,40 +460,32 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
                     &validator_metrics::ATTESTATION_SERVICE_TIMES,
                     &[validator_metrics::ATTESTATIONS_HTTP_POST],
                 );
-                if fork_name.electra_enabled() {
-                    let single_attestations = attestations
-                        .iter()
-                        .zip(validator_indices)
-                        .filter_map(|(a, i)| {
-                            match a.to_single_attestation_with_attester_index(*i) {
-                                Ok(a) => Some(a),
-                                Err(e) => {
-                                    // This shouldn't happen unless BN and VC are out of sync with
-                                    // respect to the Electra fork.
-                                    error!(
-                                        error = ?e,
-                                        committee_index = attestation_data.index,
-                                        slot = slot.as_u64(),
-                                        "type" = "unaggregated",
-                                        "Unable to convert to SingleAttestation"
-                                    );
-                                    None
-                                }
-                            }
-                        })
-                        .collect::<Vec<_>>();
 
-                    beacon_node
-                        .post_beacon_pool_attestations_v2::<S::E>(
-                            Either::Right(single_attestations),
-                            fork_name,
-                        )
-                        .await
-                } else {
-                    beacon_node
-                        .post_beacon_pool_attestations_v1(attestations)
-                        .await
-                }
+                let single_attestations = attestations
+                    .iter()
+                    .zip(validator_indices)
+                    .filter_map(|(a, i)| {
+                        match a.to_single_attestation_with_attester_index(*i) {
+                            Ok(a) => Some(a),
+                            Err(e) => {
+                                // This shouldn't happen unless BN and VC are out of sync with
+                                // respect to the Electra fork.
+                                error!(
+                                    error = ?e,
+                                    committee_index = attestation_data.index,
+                                    slot = slot.as_u64(),
+                                    "type" = "unaggregated",
+                                    "Unable to convert to SingleAttestation"
+                                );
+                                None
+                            }
+                        }
+                    })
+                    .collect::<Vec<_>>();
+
+                beacon_node
+                    .post_beacon_pool_attestations_v2::<S::E>(single_attestations, fork_name)
+                    .await
             })
             .await
         {
@@ -569,7 +560,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> AttestationService<S,
                             format!("Failed to produce an aggregate attestation: {:?}", e)
                         })?
                         .ok_or_else(|| format!("No aggregate available for {:?}", attestation_data))
-                        .map(|result| result.data)
+                        .map(|result| result.into_data())
                 } else {
                     beacon_node
                         .get_validator_aggregate_attestation_v1(

validator_client/validator_services/src/block_service.rs

@@ -1,6 +1,5 @@
 use beacon_node_fallback::{ApiTopic, BeaconNodeFallback, Error as FallbackError, Errors};
 use bls::SignatureBytes;
-use eth2::types::{FullBlockContents, PublishBlockRequest};
 use eth2::{BeaconNodeHttpClient, StatusCode};
 use graffiti_file::{determine_graffiti, GraffitiFile};
 use logging::crit;
@@ -13,11 +12,8 @@ use std::time::Duration;
 use task_executor::TaskExecutor;
 use tokio::sync::mpsc;
 use tracing::{debug, error, info, trace, warn};
-use types::{
-    BlindedBeaconBlock, BlockType, ChainSpec, EthSpec, Graffiti, PublicKeyBytes,
-    SignedBlindedBeaconBlock, Slot,
-};
-use validator_store::{Error as ValidatorStoreError, ValidatorStore};
+use types::{BlockType, ChainSpec, EthSpec, Graffiti, PublicKeyBytes, Slot};
+use validator_store::{Error as ValidatorStoreError, SignedBlock, UnsignedBlock, ValidatorStore};
 
 #[derive(Debug)]
 pub enum BlockError {
@@ -335,26 +331,10 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> BlockService<S, T> {
     ) -> Result<(), BlockError> {
         let signing_timer = validator_metrics::start_timer(&validator_metrics::BLOCK_SIGNING_TIMES);
 
-        let (block, maybe_blobs) = match unsigned_block {
-            UnsignedBlock::Full(block_contents) => {
-                let (block, maybe_blobs) = block_contents.deconstruct();
-                (block.into(), maybe_blobs)
-            }
-            UnsignedBlock::Blinded(block) => (block.into(), None),
-        };
-
         let res = self
             .validator_store
-            .sign_block(*validator_pubkey, block, slot)
-            .await
-            .map(|block| match block {
-                validator_store::SignedBlock::Full(block) => {
-                    SignedBlock::Full(PublishBlockRequest::new(Arc::new(block), maybe_blobs))
-                }
-                validator_store::SignedBlock::Blinded(block) => {
-                    SignedBlock::Blinded(Arc::new(block))
-                }
-            });
+            .sign_block(*validator_pubkey, unsigned_block, slot)
+            .await;
 
         let signed_block = match res {
             Ok(block) => block,
@@ -398,12 +378,13 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> BlockService<S, T> {
             })
             .await?;
 
+        let metadata = BlockMetadata::from(&signed_block);
         info!(
-            block_type = ?signed_block.block_type(),
-            deposits = signed_block.num_deposits(),
-            attestations = signed_block.num_attestations(),
+            block_type = ?metadata.block_type,
+            deposits = metadata.num_deposits,
+            attestations = metadata.num_attestations,
             graffiti = ?graffiti.map(|g| g.as_utf8_lossy()),
-            slot = signed_block.slot().as_u64(),
+            slot = metadata.slot.as_u64(),
             "Successfully published block"
         );
         Ok(())
@@ -508,7 +489,6 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> BlockService<S, T> {
         signed_block: &SignedBlock<S::E>,
         beacon_node: BeaconNodeHttpClient,
     ) -> Result<(), BlockError> {
-        let slot = signed_block.slot();
         match signed_block {
             SignedBlock::Full(signed_block) => {
                 let _post_timer = validator_metrics::start_timer_vec(
@@ -518,7 +498,9 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> BlockService<S, T> {
                 beacon_node
                     .post_beacon_blocks_v2_ssz(signed_block, None)
                     .await
-                    .or_else(|e| handle_block_post_error(e, slot))?
+                    .or_else(|e| {
+                        handle_block_post_error(e, signed_block.signed_block().message().slot())
+                    })?
             }
             SignedBlock::Blinded(signed_block) => {
                 let _post_timer = validator_metrics::start_timer_vec(
@@ -528,7 +510,7 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> BlockService<S, T> {
                 beacon_node
                     .post_beacon_blinded_blocks_v2_ssz(signed_block, None)
                     .await
-                    .or_else(|e| handle_block_post_error(e, slot))?
+                    .or_else(|e| handle_block_post_error(e, signed_block.message().slot()))?
             }
         }
         Ok::<_, BlockError>(())
@@ -557,13 +539,17 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> BlockService<S, T> {
                 ))
             })?;
 
-        let unsigned_block = match block_response.data {
-            eth2::types::ProduceBlockV3Response::Full(block) => UnsignedBlock::Full(block),
-            eth2::types::ProduceBlockV3Response::Blinded(block) => UnsignedBlock::Blinded(block),
+        let (block_proposer, unsigned_block) = match block_response.data {
+            eth2::types::ProduceBlockV3Response::Full(block) => {
+                (block.block().proposer_index(), UnsignedBlock::Full(block))
+            }
+            eth2::types::ProduceBlockV3Response::Blinded(block) => {
+                (block.proposer_index(), UnsignedBlock::Blinded(block))
+            }
         };
 
         info!(slot = slot.as_u64(), "Received unsigned block");
-        if proposer_index != Some(unsigned_block.proposer_index()) {
+        if proposer_index != Some(block_proposer) {
             return Err(BlockError::Recoverable(
                 "Proposer index does not match block proposer. Beacon chain re-orged".to_string(),
             ));
@@ -573,49 +559,30 @@ impl<S: ValidatorStore + 'static, T: SlotClock + 'static> BlockService<S, T> {
     }
 }
 
-pub enum UnsignedBlock<E: EthSpec> {
-    Full(FullBlockContents<E>),
-    Blinded(BlindedBeaconBlock<E>),
+/// Wrapper for values we want to log about a block we signed, for easy extraction from the possible
+/// variants.
+struct BlockMetadata {
+    block_type: BlockType,
+    slot: Slot,
+    num_deposits: usize,
+    num_attestations: usize,
 }
 
-impl<E: EthSpec> UnsignedBlock<E> {
-    pub fn proposer_index(&self) -> u64 {
-        match self {
-            UnsignedBlock::Full(block) => block.block().proposer_index(),
-            UnsignedBlock::Blinded(block) => block.proposer_index(),
-        }
-    }
-}
-
-#[derive(Debug)]
-pub enum SignedBlock<E: EthSpec> {
-    Full(PublishBlockRequest<E>),
-    Blinded(Arc<SignedBlindedBeaconBlock<E>>),
-}
-
-impl<E: EthSpec> SignedBlock<E> {
-    pub fn block_type(&self) -> BlockType {
-        match self {
-            SignedBlock::Full(_) => BlockType::Full,
-            SignedBlock::Blinded(_) => BlockType::Blinded,
-        }
-    }
-    pub fn slot(&self) -> Slot {
-        match self {
-            SignedBlock::Full(block) => block.signed_block().message().slot(),
-            SignedBlock::Blinded(block) => block.message().slot(),
-        }
-    }
-    pub fn num_deposits(&self) -> usize {
-        match self {
-            SignedBlock::Full(block) => block.signed_block().message().body().deposits().len(),
-            SignedBlock::Blinded(block) => block.message().body().deposits().len(),
-        }
-    }
-    pub fn num_attestations(&self) -> usize {
-        match self {
-            SignedBlock::Full(block) => block.signed_block().message().body().attestations_len(),
-            SignedBlock::Blinded(block) => block.message().body().attestations_len(),
+impl<E: EthSpec> From<&SignedBlock<E>> for BlockMetadata {
+    fn from(value: &SignedBlock<E>) -> Self {
+        match value {
+            SignedBlock::Full(block) => BlockMetadata {
+                block_type: BlockType::Full,
+                slot: block.signed_block().message().slot(),
+                num_deposits: block.signed_block().message().body().deposits().len(),
+                num_attestations: block.signed_block().message().body().attestations_len(),
+            },
+            SignedBlock::Blinded(block) => BlockMetadata {
+                block_type: BlockType::Blinded,
+                slot: block.message().slot(),
+                num_deposits: block.message().body().deposits().len(),
+                num_attestations: block.message().body().attestations_len(),
+            },
         }
     }
 }

validator_client/validator_services/src/latency_service.rs

@@ -0,0 +1,60 @@
+use beacon_node_fallback::BeaconNodeFallback;
+use slot_clock::SlotClock;
+use std::sync::Arc;
+use task_executor::TaskExecutor;
+use tokio::time::sleep;
+use tracing::debug;
+
+/// The latency service will run 11/12ths of the way through the slot.
+pub const SLOT_DELAY_MULTIPLIER: u32 = 11;
+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>(
+    executor: TaskExecutor,
+    slot_clock: T,
+    beacon_nodes: Arc<BeaconNodeFallback<T>>,
+) {
+    let future = async move {
+        loop {
+            let sleep_time = slot_clock
+                .duration_to_next_slot()
+                .map(|next_slot| {
+                    // This is 11/12ths through the next slot. On mainnet this
+                    // will happen in the 11th second of each slot, one second
+                    // before the next slot.
+                    next_slot + (next_slot / SLOT_DELAY_DENOMINATOR) * SLOT_DELAY_MULTIPLIER
+                })
+                // If we can't read the slot clock, just wait one slot. Running
+                // the measurement at a non-exact time is not a big issue.
+                .unwrap_or_else(|| slot_clock.slot_duration());
+
+            // Sleep until it's time to perform the measurement.
+            sleep(sleep_time).await;
+
+            for (i, measurement) in beacon_nodes.measure_latency().await.iter().enumerate() {
+                if let Some(latency) = measurement.latency {
+                    debug!(
+                        node = &measurement.beacon_node_id,
+                        latency = latency.as_millis(),
+                        "Measured BN latency"
+                    );
+                    validator_metrics::observe_timer_vec(
+                        &validator_metrics::VC_BEACON_NODE_LATENCY,
+                        &[&measurement.beacon_node_id],
+                        latency,
+                    );
+                    if i == 0 {
+                        validator_metrics::observe_duration(
+                            &validator_metrics::VC_BEACON_NODE_LATENCY_PRIMARY_ENDPOINT,
+                            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

@@ -0,0 +1,153 @@
+use crate::duties_service::DutiesService;
+use slot_clock::SlotClock;
+use std::sync::Arc;
+use task_executor::TaskExecutor;
+use tokio::time::{sleep, Duration};
+use tracing::{debug, error, info};
+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<S: ValidatorStore + 'static, T: SlotClock + 'static>(
+    duties_service: Arc<DutiesService<S, T>>,
+    executor: TaskExecutor,
+    spec: &ChainSpec,
+) -> Result<(), String> {
+    let slot_duration = Duration::from_secs(spec.seconds_per_slot);
+
+    let interval_fut = async move {
+        loop {
+            if let Some(duration_to_next_slot) = duties_service.slot_clock.duration_to_next_slot() {
+                sleep(duration_to_next_slot + slot_duration / 2).await;
+                notify(&duties_service).await;
+            } else {
+                error!("Failed to read slot clock");
+                // If we can't read the slot clock, just wait another slot.
+                sleep(slot_duration).await;
+                continue;
+            }
+        }
+    };
+
+    executor.spawn(interval_fut, "validator_notifier");
+    Ok(())
+}
+
+/// Performs a single notification routine.
+async fn notify<S: ValidatorStore, T: SlotClock + 'static>(duties_service: &DutiesService<S, T>) {
+    let (candidate_info, num_available, num_synced) =
+        duties_service.beacon_nodes.get_notifier_info().await;
+    let num_total = candidate_info.len();
+    let num_synced_fallback = num_synced.saturating_sub(1);
+
+    set_gauge(
+        &validator_metrics::AVAILABLE_BEACON_NODES_COUNT,
+        num_available as i64,
+    );
+    set_gauge(
+        &validator_metrics::SYNCED_BEACON_NODES_COUNT,
+        num_synced as i64,
+    );
+    set_gauge(
+        &validator_metrics::TOTAL_BEACON_NODES_COUNT,
+        num_total as i64,
+    );
+    if num_synced > 0 {
+        let primary = candidate_info
+            .first()
+            .map(|candidate| candidate.endpoint.as_str())
+            .unwrap_or("None");
+        info!(
+            primary,
+            total = num_total,
+            available = num_available,
+            synced = num_synced,
+            "Connected to beacon node(s)"
+        )
+    } else {
+        error!(
+            total = num_total,
+            available = num_available,
+            synced = num_synced,
+            "No synced beacon nodes"
+        )
+    }
+    if num_synced_fallback > 0 {
+        set_gauge(&validator_metrics::ETH2_FALLBACK_CONNECTED, 1);
+    } else {
+        set_gauge(&validator_metrics::ETH2_FALLBACK_CONNECTED, 0);
+    }
+
+    for info in candidate_info {
+        if let Ok(health) = info.health {
+            debug!(
+                status = "Connected",
+                index = info.index,
+                endpoint = info.endpoint,
+                head_slot = %health.head,
+                is_optimistic = ?health.optimistic_status,
+                execution_engine_status = ?health.execution_status,
+                health_tier = %health.health_tier,
+                "Beacon node info"
+            );
+        } else {
+            debug!(
+                status = "Disconnected",
+                index = info.index,
+                endpoint = info.endpoint,
+                "Beacon node info"
+            );
+        }
+    }
+
+    if let Some(slot) = duties_service.slot_clock.now() {
+        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);
+        let attesting_validators = duties_service.attester_count(epoch);
+        let doppelganger_detecting_validators = duties_service.doppelganger_detecting_count();
+
+        if doppelganger_detecting_validators > 0 {
+            info!(
+                doppelganger_detecting_validators,
+                "Listening for doppelgangers"
+            )
+        }
+
+        if total_validators == 0 {
+            info!(
+                msg = "see `lighthouse vm create --help` or the HTTP API documentation",
+                "No validators present"
+            )
+        } else if total_validators == attesting_validators {
+            info!(
+                current_epoch_proposers = proposing_validators,
+                active_validators = attesting_validators,
+                total_validators = total_validators,
+                %epoch,
+                %slot,
+                "All validators active"
+            );
+        } else if attesting_validators > 0 {
+            info!(
+                current_epoch_proposers = proposing_validators,
+                active_validators = attesting_validators,
+                total_validators = total_validators,
+                %epoch,
+                %slot,
+                "Some validators active"
+            );
+        } else {
+            info!(
+                validators = total_validators,
+                %epoch,
+                %slot,
+                "Awaiting activation"
+            );
+        }
+    } else {
+        error!("Unable to read slot clock");
+    }
+}

validator_client/validator_services/src/sync.rs

@@ -595,8 +595,12 @@ pub async fn fill_in_aggregation_proofs<S: ValidatorStore, T: SlotClock + 'stati
     current_slot: Slot,
     pre_compute_slot: Slot,
 ) {
+    // Start at the next slot, as aggregation proofs for the duty at the current slot are no longer
+    // required since we do the actual aggregation in the slot before the duty slot.
+    let start_slot = current_slot.as_u64() + 1;
+
     // Generate selection proofs for each validator at each slot, one slot at a time.
-    for slot in (current_slot.as_u64()..=pre_compute_slot.as_u64()).map(Slot::new) {
+    for slot in (start_slot..=pre_compute_slot.as_u64()).map(Slot::new) {
         // For distributed mode
         if duties_service
             .sync_duties