Anchor pre-PR: Modularize validator store (#6771)

* pass slots_per_epoch at runtime

* remove generic E from unrequired types

* move `validator_store` to `lighthouse_validator_store`

* make validator_store into a trait

* further reduce dependencies

* remove `environment` dependency on `beacon_node_fallback`

* Manually pull in some changes from tracing-integration (thanks sayan!)

Co-authored-by: ThreeHrSleep <threehrsleep@gmail.com>

* remove `environment` from `validator_services`

* unify boost factor accessors

* add builder for DutiesService

* Manually merge tracing PR for beacon_node_fallback

Co-authored-by: ThreeHrSleep <threehrsleep@gmail.com>

* Fix chain_spec for BlockService

* address review

* remove PhantomData from SyncDutiesMap

* fix tests

* correct test

* Add `E` to `ValidatorStore` as associated type

* fix tests

* derive Clone for ValidatorStore's Error and required sub-errors

* switch to enum for block signing to allow differing types

---------

Co-authored-by: João Oliveira <hello@jxs.pt>
Co-authored-by: ThreeHrSleep <threehrsleep@gmail.com>
Co-authored-by: Jimmy Chen <jimmy@sigmaprime.io>

Cargo.toml

@@ -89,11 +89,11 @@ members = [
     "validator_client/http_api",
     "validator_client/http_metrics",
     "validator_client/initialized_validators",
+    "validator_client/lighthouse_validator_store",
     "validator_client/signing_method",
     "validator_client/slashing_protection",
     "validator_client/validator_metrics",
     "validator_client/validator_services",
-    "validator_client/validator_store",
 
     "validator_manager",
 
@@ -253,6 +253,7 @@ int_to_bytes = { path = "consensus/int_to_bytes" }
 kzg = { path = "crypto/kzg" }
 metrics = { path = "common/metrics" }
 lighthouse_network = { path = "beacon_node/lighthouse_network" }
+lighthouse_validator_store = { path = "validator_client/lighthouse_validator_store" }
 lighthouse_version = { path = "common/lighthouse_version" }
 lockfile = { path = "common/lockfile" }
 logging = { path = "common/logging" }

common/logging/src/lib.rs

@@ -13,6 +13,7 @@ use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 pub const MAX_MESSAGE_WIDTH: usize = 40;
 
 pub mod async_record;
+pub mod macros;
 mod sse_logging_components;
 mod tracing_logging_layer;
 mod tracing_metrics_layer;

common/logging/src/macros.rs

@@ -0,0 +1,6 @@
+#[macro_export]
+macro_rules! crit {
+    ($($arg:tt)*) => {
+        tracing::error!(error_type = "crit",  $($arg)*);
+    };
+}

consensus/types/src/attestation.rs

@@ -16,7 +16,7 @@ use super::{
     Signature, SignedRoot,
 };
 
-#[derive(Debug, PartialEq)]
+#[derive(Debug, PartialEq, Clone)]
 pub enum Error {
     SszTypesError(ssz_types::Error),
     AlreadySigned(usize),

consensus/types/src/payload.rs

@@ -84,6 +84,7 @@ pub trait AbstractExecPayload<E: EthSpec>:
     + TryInto<Self::Capella>
     + TryInto<Self::Deneb>
     + TryInto<Self::Electra>
+    + Sync
 {
     type Ref<'a>: ExecPayload<E>
         + Copy
@@ -95,19 +96,23 @@ pub trait AbstractExecPayload<E: EthSpec>:
     type Bellatrix: OwnedExecPayload<E>
         + Into<Self>
         + for<'a> From<Cow<'a, ExecutionPayloadBellatrix<E>>>
-        + TryFrom<ExecutionPayloadHeaderBellatrix<E>>;
+        + TryFrom<ExecutionPayloadHeaderBellatrix<E>>
+        + Sync;
     type Capella: OwnedExecPayload<E>
         + Into<Self>
         + for<'a> From<Cow<'a, ExecutionPayloadCapella<E>>>
-        + TryFrom<ExecutionPayloadHeaderCapella<E>>;
+        + TryFrom<ExecutionPayloadHeaderCapella<E>>
+        + Sync;
     type Deneb: OwnedExecPayload<E>
         + Into<Self>
         + for<'a> From<Cow<'a, ExecutionPayloadDeneb<E>>>
-        + TryFrom<ExecutionPayloadHeaderDeneb<E>>;
+        + TryFrom<ExecutionPayloadHeaderDeneb<E>>
+        + Sync;
     type Electra: OwnedExecPayload<E>
         + Into<Self>
         + for<'a> From<Cow<'a, ExecutionPayloadElectra<E>>>
-        + TryFrom<ExecutionPayloadHeaderElectra<E>>;
+        + TryFrom<ExecutionPayloadHeaderElectra<E>>
+        + Sync;
 }
 
 #[superstruct(

testing/web3signer_tests/Cargo.toml

@@ -14,6 +14,7 @@ eth2_keystore = { workspace = true }
 eth2_network_config = { workspace = true }
 futures = { workspace = true }
 initialized_validators = { workspace = true }
+lighthouse_validator_store = { workspace = true }
 logging = { workspace = true }
 parking_lot = { workspace = true }
 reqwest = { workspace = true }

testing/web3signer_tests/src/lib.rs

@@ -25,6 +25,7 @@ mod tests {
     use initialized_validators::{
         load_pem_certificate, load_pkcs12_identity, InitializedValidators,
     };
+    use lighthouse_validator_store::LighthouseValidatorStore;
     use logging::test_logger;
     use parking_lot::Mutex;
     use reqwest::Client;
@@ -45,7 +46,7 @@ mod tests {
     use tokio::time::sleep;
     use types::{attestation::AttestationBase, *};
     use url::Url;
-    use validator_store::{Error as ValidatorStoreError, ValidatorStore};
+    use validator_store::{Error as ValidatorStoreError, SignBlock, ValidatorStore};
 
     /// If the we are unable to reach the Web3Signer HTTP API within this time out then we will
     /// assume it failed to start.
@@ -309,7 +310,7 @@ mod tests {
 
     /// A testing rig which holds a `ValidatorStore`.
     struct ValidatorStoreRig {
-        validator_store: Arc<ValidatorStore<TestingSlotClock, E>>,
+        validator_store: Arc<LighthouseValidatorStore<TestingSlotClock, E>>,
         _validator_dir: TempDir,
         runtime: Arc<tokio::runtime::Runtime>,
         _runtime_shutdown: async_channel::Sender<()>,
@@ -358,12 +359,12 @@ mod tests {
 
             let slot_clock =
                 TestingSlotClock::new(Slot::new(0), Duration::from_secs(0), Duration::from_secs(1));
-            let config = validator_store::Config {
+            let config = lighthouse_validator_store::Config {
                 enable_web3signer_slashing_protection: slashing_protection_config.local,
                 ..Default::default()
             };
 
-            let validator_store = ValidatorStore::<_, E>::new(
+            let validator_store = LighthouseValidatorStore::new(
                 initialized_validators,
                 slashing_protection,
                 Hash256::repeat_byte(42),
@@ -488,7 +489,7 @@ mod tests {
             generate_sig: F,
         ) -> Self
         where
-            F: Fn(PublicKeyBytes, Arc<ValidatorStore<TestingSlotClock, E>>) -> R,
+            F: Fn(PublicKeyBytes, Arc<LighthouseValidatorStore<TestingSlotClock, E>>) -> R,
             R: Future<Output = S>,
             // We use the `SignedObject` trait to white-list objects for comparison. This avoids
             // accidentally comparing something meaningless like a `()`.
@@ -523,8 +524,8 @@ mod tests {
             web3signer_should_sign: bool,
         ) -> Self
         where
-            F: Fn(PublicKeyBytes, Arc<ValidatorStore<TestingSlotClock, E>>) -> R,
-            R: Future<Output = Result<(), ValidatorStoreError>>,
+            F: Fn(PublicKeyBytes, Arc<LighthouseValidatorStore<TestingSlotClock, E>>) -> R,
+            R: Future<Output = Result<(), lighthouse_validator_store::Error>>,
         {
             for validator_rig in &self.validator_rigs {
                 let result =

validator_client/Cargo.toml

@@ -25,6 +25,7 @@ fdlimit = "0.3.0"
 graffiti_file = { workspace = true }
 hyper = { workspace = true }
 initialized_validators = { workspace = true }
+lighthouse_validator_store = { workspace = true }
 metrics = { workspace = true }
 monitoring_api = { workspace = true }
 parking_lot = { workspace = true }

validator_client/beacon_node_fallback/Cargo.toml

@@ -9,14 +9,14 @@ name = "beacon_node_fallback"
 path = "src/lib.rs"
 
 [dependencies]
-environment = { workspace = true }
 eth2 = { workspace = true }
 futures = { workspace = true }
 itertools = { workspace = true }
 serde = { workspace = true }
-slog = { workspace = true }
 slot_clock = { workspace = true }
 strum = { workspace = true }
+task_executor = { workspace = true }
 tokio = { workspace = true }
+tracing = { workspace = true }
 types = { workspace = true }
 validator_metrics = { workspace = true }

validator_client/beacon_node_fallback/src/beacon_node_health.rs

@@ -2,10 +2,10 @@ use super::CandidateError;
 use eth2::BeaconNodeHttpClient;
 use itertools::Itertools;
 use serde::{Deserialize, Serialize};
-use slog::{warn, Logger};
 use std::cmp::Ordering;
 use std::fmt::{Debug, Display, Formatter};
 use std::str::FromStr;
+use tracing::warn;
 use types::Slot;
 
 /// Sync distances between 0 and DEFAULT_SYNC_TOLERANCE are considered `synced`.
@@ -290,15 +290,13 @@ impl BeaconNodeHealth {
 
 pub async fn check_node_health(
     beacon_node: &BeaconNodeHttpClient,
-    log: &Logger,
 ) -> Result<(Slot, bool, bool), CandidateError> {
     let resp = match beacon_node.get_node_syncing().await {
         Ok(resp) => resp,
         Err(e) => {
             warn!(
-                log,
-                "Unable connect to beacon node";
-                "error" => %e
+                error = %e,
+                "Unable connect to beacon node"
             );
 
             return Err(CandidateError::Offline);