Merge remote-tracking branch 'origin/release-v8.1' into back-merge-v8.1.1

Cargo.lock

@@ -4,7 +4,7 @@ version = 4
 
 [[package]]
 name = "account_manager"
-version = "8.1.0"
+version = "8.1.1"
 dependencies = [
  "account_utils",
  "bls",
@@ -1285,7 +1285,7 @@ dependencies = [
 
 [[package]]
 name = "beacon_node"
-version = "8.1.0"
+version = "8.1.1"
 dependencies = [
  "account_utils",
  "beacon_chain",
@@ -1548,7 +1548,7 @@ dependencies = [
 
 [[package]]
 name = "boot_node"
-version = "8.1.0"
+version = "8.1.1"
 dependencies = [
  "beacon_node",
  "bytes",
@@ -4802,9 +4802,9 @@ dependencies = [
 
 [[package]]
 name = "keccak"
-version = "0.1.5"
+version = "0.1.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ecc2af9a1119c51f12a14607e783cb977bde58bc069ff0c3da1095e635d70654"
+checksum = "cb26cec98cce3a3d96cbb7bced3c4b16e3d13f27ec56dbd62cbc8f39cfb9d653"
 dependencies = [
  "cpufeatures",
 ]
@@ -4867,7 +4867,7 @@ checksum = "830d08ce1d1d941e6b30645f1a0eb5643013d835ce3779a5fc208261dbe10f55"
 
 [[package]]
 name = "lcli"
-version = "8.1.0"
+version = "8.1.1"
 dependencies = [
  "account_utils",
  "beacon_chain",
@@ -5353,7 +5353,7 @@ dependencies = [
 
 [[package]]
 name = "lighthouse"
-version = "8.1.0"
+version = "8.1.1"
 dependencies = [
  "account_manager",
  "account_utils",
@@ -5485,7 +5485,7 @@ dependencies = [
 
 [[package]]
 name = "lighthouse_version"
-version = "8.1.0"
+version = "8.1.1"
 dependencies = [
  "regex",
 ]
@@ -9534,7 +9534,7 @@ dependencies = [
 
 [[package]]
 name = "validator_client"
-version = "8.1.0"
+version = "8.1.1"
 dependencies = [
  "account_utils",
  "beacon_node_fallback",
@@ -10521,8 +10521,7 @@ dependencies = [
 [[package]]
 name = "yamux"
 version = "0.13.8"
-source = "registry+https://github.com/rust-lang/crates.io-index"
+source = "git+https://github.com/sigp/rust-yamux?rev=575b17c0f44f4253079a6bafaa2de74ca1d6dfaa#575b17c0f44f4253079a6bafaa2de74ca1d6dfaa"
-checksum = "deab71f2e20691b4728b349c6cee8fc7223880fa67b6b4f92225ec32225447e5"
 dependencies = [
  "futures",
  "log",

Cargo.toml

@@ -91,7 +91,7 @@ resolver = "2"
 
 [workspace.package]
 edition = "2024"
-version = "8.1.0"
+version = "8.1.1"
 
 [workspace.dependencies]
 account_utils = { path = "common/account_utils" }
@@ -302,3 +302,4 @@ debug = true
 
 [patch.crates-io]
 quick-protobuf = { git = "https://github.com/sigp/quick-protobuf.git", rev = "681f413312404ab6e51f0b46f39b0075c6f4ebfd" }
+yamux = { git = "https://github.com/sigp/rust-yamux", rev = "575b17c0f44f4253079a6bafaa2de74ca1d6dfaa" }

beacon_node/network/src/metrics.rs

@@ -462,6 +462,13 @@ pub static SYNCING_CHAIN_BATCH_AWAITING_PROCESSING: LazyLock<Result<Histogram>>
             ]),
         )
     });
+pub static SYNCING_CHAIN_BATCHES: LazyLock<Result<IntGaugeVec>> = LazyLock::new(|| {
+    try_create_int_gauge_vec(
+        "sync_batches",
+        "Number of batches in sync chains by sync type and state",
+        &["sync_type", "state"],
+    )
+});
 pub static SYNC_SINGLE_BLOCK_LOOKUPS: LazyLock<Result<IntGauge>> = LazyLock::new(|| {
     try_create_int_gauge(
         "sync_single_block_lookups",

beacon_node/network/src/sync/backfill_sync/mod.rs

@@ -8,9 +8,11 @@
 //! If a batch fails, the backfill sync cannot progress. In this scenario, we mark the backfill
 //! sync as failed, log an error and attempt to retry once a new peer joins the node.
 
+use crate::metrics;
 use crate::network_beacon_processor::ChainSegmentProcessId;
 use crate::sync::batch::{
-    BatchConfig, BatchId, BatchInfo, BatchOperationOutcome, BatchProcessingResult, BatchState,
+    BatchConfig, BatchId, BatchInfo, BatchMetricsState, BatchOperationOutcome,
+    BatchProcessingResult, BatchState,
 };
 use crate::sync::block_sidecar_coupling::CouplingError;
 use crate::sync::manager::BatchProcessResult;
@@ -31,6 +33,7 @@ use std::collections::{
 use std::hash::{Hash, Hasher};
 use std::marker::PhantomData;
 use std::sync::Arc;
+use strum::IntoEnumIterator;
 use tracing::{debug, error, info, warn};
 use types::{ColumnIndex, Epoch, EthSpec};
 
@@ -1181,6 +1184,21 @@ impl<T: BeaconChainTypes> BackFillSync<T> {
                 .epoch(T::EthSpec::slots_per_epoch())
     }
 
+    pub fn register_metrics(&self) {
+        for state in BatchMetricsState::iter() {
+            let count = self
+                .batches
+                .values()
+                .filter(|b| b.state().metrics_state() == state)
+                .count();
+            metrics::set_gauge_vec(
+                &metrics::SYNCING_CHAIN_BATCHES,
+                &["backfill", state.into()],
+                count as i64,
+            );
+        }
+    }
+
     /// Updates the global network state indicating the current state of a backfill sync.
     fn set_state(&self, state: BackFillState) {
         *self.network_globals.backfill_state.write() = state;

beacon_node/network/src/sync/batch.rs

@@ -10,10 +10,22 @@ use std::marker::PhantomData;
 use std::ops::Sub;
 use std::time::Duration;
 use std::time::Instant;
-use strum::Display;
+use strum::{Display, EnumIter, IntoStaticStr};
 use types::Slot;
 use types::{DataColumnSidecarList, Epoch, EthSpec};
 
+/// Batch states used as metrics labels.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, EnumIter, IntoStaticStr)]
+#[strum(serialize_all = "snake_case")]
+pub enum BatchMetricsState {
+    AwaitingDownload,
+    Downloading,
+    AwaitingProcessing,
+    Processing,
+    AwaitingValidation,
+    Failed,
+}
+
 pub type BatchId = Epoch;
 
 /// Type of expected batch.
@@ -142,6 +154,18 @@ impl<D: Hash> BatchState<D> {
     pub fn poison(&mut self) -> BatchState<D> {
         std::mem::replace(self, BatchState::Poisoned)
     }
+
+    /// Returns the metrics state for this batch.
+    pub fn metrics_state(&self) -> BatchMetricsState {
+        match self {
+            BatchState::AwaitingDownload => BatchMetricsState::AwaitingDownload,
+            BatchState::Downloading(_) => BatchMetricsState::Downloading,
+            BatchState::AwaitingProcessing(..) => BatchMetricsState::AwaitingProcessing,
+            BatchState::Processing(_) => BatchMetricsState::Processing,
+            BatchState::AwaitingValidation(_) => BatchMetricsState::AwaitingValidation,
+            BatchState::Poisoned | BatchState::Failed => BatchMetricsState::Failed,
+        }
+    }
 }
 
 impl<E: EthSpec, B: BatchConfig, D: Hash> BatchInfo<E, B, D> {

beacon_node/network/src/sync/custody_backfill_sync/mod.rs

@@ -12,14 +12,16 @@ use lighthouse_network::{
 };
 use logging::crit;
 use std::hash::{DefaultHasher, Hash, Hasher};
+use strum::IntoEnumIterator;
 use tracing::{debug, error, info, info_span, warn};
 use types::{DataColumnSidecarList, Epoch, EthSpec};
 
+use crate::metrics;
 use crate::sync::{
     backfill_sync::{BACKFILL_EPOCHS_PER_BATCH, ProcessResult, SyncStart},
     batch::{
-        BatchConfig, BatchId, BatchInfo, BatchOperationOutcome, BatchProcessingResult, BatchState,
+        BatchConfig, BatchId, BatchInfo, BatchMetricsState, BatchOperationOutcome,
-        ByRangeRequestType,
+        BatchProcessingResult, BatchState, ByRangeRequestType,
     },
     block_sidecar_coupling::CouplingError,
     manager::CustodyBatchProcessResult,
@@ -1114,6 +1116,21 @@ impl<T: BeaconChainTypes> CustodyBackFillSync<T> {
         *self.network_globals.custody_sync_state.write() = state;
     }
 
+    pub fn register_metrics(&self) {
+        for state in BatchMetricsState::iter() {
+            let count = self
+                .batches
+                .values()
+                .filter(|b| b.state().metrics_state() == state)
+                .count();
+            metrics::set_gauge_vec(
+                &metrics::SYNCING_CHAIN_BATCHES,
+                &["custody_backfill", state.into()],
+                count as i64,
+            );
+        }
+    }
+
     /// A fully synced peer has joined us.
     /// If we are in a failed state, update a local variable to indicate we are able to restart
     /// the failed sync on the next attempt.

beacon_node/network/src/sync/manager.rs

@@ -787,6 +787,9 @@ impl<T: BeaconChainTypes> SyncManager<T> {
                 }
                 _ = register_metrics_interval.tick() => {
                     self.network.register_metrics();
+                    self.range_sync.register_metrics();
+                    self.backfill_sync.register_metrics();
+                    self.custody_backfill_sync.register_metrics();
                 }
                 _ = epoch_interval.tick() => {
                     self.update_sync_state();

beacon_node/network/src/sync/range_sync/chain.rs

@@ -3,7 +3,8 @@ use crate::metrics;
 use crate::network_beacon_processor::ChainSegmentProcessId;
 use crate::sync::batch::BatchId;
 use crate::sync::batch::{
-    BatchConfig, BatchInfo, BatchOperationOutcome, BatchProcessingResult, BatchState,
+    BatchConfig, BatchInfo, BatchMetricsState, BatchOperationOutcome, BatchProcessingResult,
+    BatchState,
 };
 use crate::sync::block_sidecar_coupling::CouplingError;
 use crate::sync::network_context::{RangeRequestId, RpcRequestSendError, RpcResponseError};
@@ -234,6 +235,14 @@ impl<T: BeaconChainTypes> SyncingChain<T> {
             .sum()
     }
 
+    /// Returns the number of batches in the given metrics state.
+    pub fn count_batches_in_state(&self, state: BatchMetricsState) -> usize {
+        self.batches
+            .values()
+            .filter(|b| b.state().metrics_state() == state)
+            .count()
+    }
+
     /// Removes a peer from the chain.
     /// If the peer has active batches, those are considered failed and re-requested.
     pub fn remove_peer(&mut self, peer_id: &PeerId) -> ProcessingResult {

beacon_node/network/src/sync/range_sync/chain_collection.rs

@@ -6,6 +6,7 @@
 use super::chain::{ChainId, ProcessingResult, RemoveChain, SyncingChain};
 use super::sync_type::RangeSyncType;
 use crate::metrics;
+use crate::sync::batch::BatchMetricsState;
 use crate::sync::network_context::SyncNetworkContext;
 use beacon_chain::{BeaconChain, BeaconChainTypes};
 use fnv::FnvHashMap;
@@ -17,6 +18,7 @@ use smallvec::SmallVec;
 use std::collections::HashMap;
 use std::collections::hash_map::Entry;
 use std::sync::Arc;
+use strum::IntoEnumIterator;
 use tracing::{debug, error};
 use types::EthSpec;
 use types::{Epoch, Hash256, Slot};
@@ -372,7 +374,8 @@ impl<T: BeaconChainTypes> ChainCollection<T> {
             .iter()
             .map(|(id, chain)| (chain.available_peers(), !chain.is_syncing(), *id))
             .collect::<Vec<_>>();
-        preferred_ids.sort_unstable();
+        // Sort in descending order
+        preferred_ids.sort_unstable_by(|a, b| b.cmp(a));
 
         let mut syncing_chains = SmallVec::<[Id; PARALLEL_HEAD_CHAINS]>::new();
         for (_, _, id) in preferred_ids {
@@ -540,6 +543,25 @@ impl<T: BeaconChainTypes> ChainCollection<T> {
         }
     }
 
+    pub fn register_metrics(&self) {
+        for (sync_type, chains) in [
+            ("range_finalized", &self.finalized_chains),
+            ("range_head", &self.head_chains),
+        ] {
+            for state in BatchMetricsState::iter() {
+                let count: usize = chains
+                    .values()
+                    .map(|chain| chain.count_batches_in_state(state))
+                    .sum();
+                metrics::set_gauge_vec(
+                    &metrics::SYNCING_CHAIN_BATCHES,
+                    &[sync_type, state.into()],
+                    count as i64,
+                );
+            }
+        }
+    }
+
     fn update_metrics(&self) {
         metrics::set_gauge_vec(
             &metrics::SYNCING_CHAINS_COUNT,

beacon_node/network/src/sync/range_sync/range.rs

@@ -376,6 +376,10 @@ where
             .update(network, &local, &mut self.awaiting_head_peers);
     }
 
+    pub fn register_metrics(&self) {
+        self.chains.register_metrics();
+    }
+
     /// Kickstarts sync.
     pub fn resume(&mut self, network: &mut SyncNetworkContext<T>) {
         for (removed_chain, sync_type, remove_reason) in