gnosis/lighthouse
1
0
Fork 0
mirror of https://github.com/sigp/lighthouse.git synced 2026-04-30 19:23:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'unstable' into vc-fallback

Browse Source
This commit is contained in:
Mac L
2023-11-02 16:27:14 +11:00
parent 23f68ad61c 42da392edc
commit 235c4524f7
546 changed files with 32366 additions and 8974 deletions
Download Patch File Download Diff File Expand all files Collapse all files

101
validator_client/Cargo.toml
View File

@@ -2,63 +2,62 @@
name = "validator_client"
version = "0.3.5"
authors = ["Paul Hauner <paul@paulhauner.com>", "Age Manning <Age@AgeManning.com>", "Luke Anderson <luke@lukeanderson.com.au>"]
edition = "2021"
edition = { workspace = true }
[lib]
name = "validator_client"
path = "src/lib.rs"
[dev-dependencies]
tokio = { version = "1.14.0", features = ["time", "rt-multi-thread", "macros"] }
tokio = { workspace = true }
[dependencies]
tree_hash = "0.5.0"
clap = "2.33.3"
slashing_protection = { path = "./slashing_protection" }
slot_clock = { path = "../common/slot_clock" }
types = { path = "../consensus/types" }
safe_arith = { path = "../consensus/safe_arith" }
serde = "1.0.116"
serde_derive = "1.0.116"
bincode = "1.3.1"
serde_json = "1.0.58"
slog = { version = "2.5.2", features = ["max_level_trace", "release_max_level_trace"] }
tokio = { version = "1.14.0", features = ["time"] }
tokio-stream = { version = "0.1.3", features = ["sync"] }
futures = "0.3.7"
dirs = "3.0.1"
directory = { path = "../common/directory" }
lockfile = { path = "../common/lockfile" }
environment = { path = "../lighthouse/environment" }
parking_lot = "0.12.0"
exit-future = "0.2.0"
filesystem = { path = "../common/filesystem" }
hex = "0.4.2"
deposit_contract = { path = "../common/deposit_contract" }
bls = { path = "../crypto/bls" }
eth2 = { path = "../common/eth2" }
tempfile = "3.1.0"
validator_dir = { path = "../common/validator_dir" }
clap_utils = { path = "../common/clap_utils" }
eth2_keystore = { path = "../crypto/eth2_keystore" }
account_utils = { path = "../common/account_utils" }
lighthouse_version = { path = "../common/lighthouse_version" }
warp_utils = { path = "../common/warp_utils" }
warp = "0.3.2"
hyper = "0.14.4"
ethereum_serde_utils = "0.5.0"
libsecp256k1 = "0.7.0"
ring = "0.16.19"
rand = { version = "0.8.5", features = ["small_rng"] }
lighthouse_metrics = { path = "../common/lighthouse_metrics" }
lazy_static = "1.4.0"
itertools = "0.10.0"
monitoring_api = { path = "../common/monitoring_api" }
sensitive_url = { path = "../common/sensitive_url" }
task_executor = { path = "../common/task_executor" }
reqwest = { version = "0.11.0", features = ["json","stream"] }
url = "2.2.2"
malloc_utils = { path = "../common/malloc_utils" }
sysinfo = "0.26.5"
tree_hash = { workspace = true }
clap = { workspace = true }
slashing_protection = { workspace = true }
slot_clock = { workspace = true }
types = { workspace = true }
safe_arith = { workspace = true }
serde = { workspace = true }
bincode = { workspace = true }
serde_json = { workspace = true }
slog = { workspace = true }
tokio = { workspace = true }
tokio-stream = { workspace = true }
futures = { workspace = true }
dirs = { workspace = true }
directory = { workspace = true }
lockfile = { workspace = true }
environment = { workspace = true }
parking_lot = { workspace = true }
exit-future = { workspace = true }
filesystem = { workspace = true }
hex = { workspace = true }
deposit_contract = { workspace = true }
bls = { workspace = true }
eth2 = { workspace = true }
tempfile = { workspace = true }
validator_dir = { workspace = true }
clap_utils = { workspace = true }
eth2_keystore = { workspace = true }
account_utils = { workspace = true }
lighthouse_version = { workspace = true }
warp_utils = { workspace = true }
warp = { workspace = true }
hyper = { workspace = true }
ethereum_serde_utils = { workspace = true }
libsecp256k1 = { workspace = true }
ring = { workspace = true }
rand = { workspace = true, features = ["small_rng"] }
lighthouse_metrics = { workspace = true }
lazy_static = { workspace = true }
itertools = { workspace = true }
monitoring_api = { workspace = true }
sensitive_url = { workspace = true }
task_executor = { workspace = true }
reqwest = { workspace = true, features = ["native-tls"] }
url = { workspace = true }
malloc_utils = { workspace = true }
sysinfo = { workspace = true }
system_health = { path = "../common/system_health" }
logging = { path = "../common/logging" }
logging = { workspace = true }

28
validator_client/slashing_protection/Cargo.toml
View File

@@ -2,7 +2,7 @@
name = "slashing_protection"
version = "0.1.0"
authors = ["Michael Sproul <michael@sigmaprime.io>", "pscott <scottpiriou@gmail.com>"]
edition = "2021"
edition = { workspace = true }
autotests = false
[[test]]
@@ -10,21 +10,21 @@ name = "slashing_protection_tests"
path = "tests/main.rs"
[dependencies]
tempfile = "3.1.0"
types = { path = "../../consensus/types" }
rusqlite = { version = "0.28.0", features = ["bundled"] }
r2d2 = "0.8.9"
tempfile = { workspace = true }
types = { workspace = true }
rusqlite = { workspace = true }
r2d2 = { workspace = true }
r2d2_sqlite = "0.21.0"
serde = "1.0.116"
serde_derive = "1.0.116"
serde_json = "1.0.58"
ethereum_serde_utils = "0.5.0"
filesystem = { path = "../../common/filesystem" }
arbitrary = { version = "1.0", features = ["derive"], optional = true }
serde = { workspace = true }
serde_json = { workspace = true }
ethereum_serde_utils = { workspace = true }
filesystem = { workspace = true }
arbitrary = { workspace = true, features = ["derive"] }
[dev-dependencies]
lazy_static = "1.4.0"
rayon = "1.4.1"
lazy_static = { workspace = true }
rayon = { workspace = true }
[features]
arbitrary-fuzz = ["arbitrary", "types/arbitrary-fuzz"]
arbitrary-fuzz = ["types/arbitrary-fuzz"]
portable = ["types/portable"]

17
validator_client/slashing_protection/Makefile
View File

@@ -6,20 +6,23 @@ ARCHIVE_URL := https://github.com/eth-clients/slashing-protection-interchange-te
ifeq ($(OS),Windows_NT)
ifeq (, $(shell where rm))
rmfile = if exist $(1) (del /F /Q $(1))
rmdir = if exist $(1) (rmdir /Q /S $(1))
rmfile = if exist $(1) (del /F /Q $(1))
rmdir = if exist $(1) (rmdir /Q /S $(1))
makedir = if not exist $(1) (mkdir $(1))
else
rmfile = rm -f $(1)
rmdir = rm -rf $(1)
rmfile = rm -f $(1)
rmdir = rm -rf $(1)
makedir = mkdir -p $(1)
endif
else
rmfile = rm -f $(1)
rmdir = rm -rf $(1)
rmfile = rm -f $(1)
rmdir = rm -rf $(1)
makedir = mkdir -p $(1)
endif
$(OUTPUT_DIR): $(TARBALL)
$(call rmdir,$@)
mkdir $@
$(call makedir,$@)
tar --strip-components=1 -xzf $^ -C $@
$(TARBALL):

2
validator_client/slashing_protection/src/interchange.rs
View File

@@ -1,5 +1,5 @@
use crate::InterchangeError;
use serde_derive::{Deserialize, Serialize};
use serde::{Deserialize, Serialize};
use std::cmp::max;
use std::collections::{HashMap, HashSet};
use std::io;

2
validator_client/slashing_protection/src/interchange_test.rs
View File

@@ -3,7 +3,7 @@ use crate::{
test_utils::{pubkey, DEFAULT_GENESIS_VALIDATORS_ROOT},
SigningRoot, SlashingDatabase,
};
use serde_derive::{Deserialize, Serialize};
use serde::{Deserialize, Serialize};
use std::collections::HashSet;
use tempfile::tempdir;
use types::{Epoch, Hash256, PublicKeyBytes, Slot};

7
validator_client/slashing_protection/tests/interop.rs
View File

@@ -25,8 +25,10 @@ fn test_root_dir() -> PathBuf {
.join("tests")
}
// NOTE: I've combined two tests together to avoid a race-condition which occurs when fighting over
// which test builds the TEST_ROOT_DIR lazy static.
#[test]
fn generated() {
fn generated_and_with_minification() {
for entry in TEST_ROOT_DIR
.join("generated")
.read_dir()
@@ -37,10 +39,7 @@ fn generated() {
let test_case: MultiTestCase = serde_json::from_reader(&file).unwrap();
test_case.run(false);
}
}
#[test]
fn generated_with_minification() {
for entry in TEST_ROOT_DIR
.join("generated")
.read_dir()

10
validator_client/src/attestation_service.rs
View File

@@ -192,7 +192,7 @@ impl<T: SlotClock + 'static, E: EthSpec> AttestationService<T, E> {
.into_iter()
.fold(HashMap::new(), |mut map, duty_and_proof| {
map.entry(duty_and_proof.duty.committee_index)
.or_insert_with(Vec::new)
.or_default()
.push(duty_and_proof);
map
});
@@ -482,6 +482,14 @@ impl<T: SlotClock + 'static, E: EthSpec> AttestationService<T, E> {
) -> Result<(), String> {
let log = self.context.log();
if !validator_duties
.iter()
.any(|duty_and_proof| duty_and_proof.selection_proof.is_some())
{
// Exit early if no validator is aggregator
return Ok(());
}
let aggregated_attestation = &self
.beacon_nodes
.first_success(|beacon_node| async move {

10
validator_client/src/beacon_node_fallback.rs
View File

@@ -12,7 +12,7 @@ use environment::RuntimeContext;
use eth2::BeaconNodeHttpClient;
use futures::future;
use parking_lot::RwLock as PLRwLock;
use serde_derive::{Deserialize, Serialize};
use serde::{Deserialize, Serialize};
use slog::{debug, error, warn, Logger};
use slot_clock::SlotClock;
use std::cmp::Ordering;
@@ -322,6 +322,14 @@ impl<E: EthSpec> CandidateBeaconNode<E> {
"endpoint_capella_fork_epoch" => ?beacon_node_spec.capella_fork_epoch,
"hint" => UPDATE_REQUIRED_LOG_HINT,
);
} else if beacon_node_spec.deneb_fork_epoch != spec.deneb_fork_epoch {
warn!(
log,
"Beacon node has mismatched Deneb fork epoch";
"endpoint" => %self.beacon_node,
"endpoint_deneb_fork_epoch" => ?beacon_node_spec.deneb_fork_epoch,
"hint" => UPDATE_REQUIRED_LOG_HINT,
);
}
Ok(())

2
validator_client/src/beacon_node_health.rs
View File

@@ -1,5 +1,5 @@
use crate::beacon_node_fallback::Config;
use serde_derive::{Deserialize, Serialize};
use serde::{Deserialize, Serialize};
use std::cmp::Ordering;
use std::fmt::{Debug, Display, Formatter};
use types::Slot;

270
validator_client/src/block_service.rs
View File

@@ -6,9 +6,11 @@ use crate::{
http_metrics::metrics,
validator_store::{Error as ValidatorStoreError, ValidatorStore},
};
use bls::SignatureBytes;
use environment::RuntimeContext;
use eth2::BeaconNodeHttpClient;
use slog::{crit, debug, error, info, trace, warn};
use eth2::types::{BlockContents, SignedBlockContents};
use eth2::{BeaconNodeHttpClient, StatusCode};
use slog::{crit, debug, error, info, trace, warn, Logger};
use slot_clock::SlotClock;
use std::fmt::Debug;
use std::future::Future;
@@ -454,68 +456,23 @@ impl<T: SlotClock + 'static, E: EthSpec> BlockService<T, E> {
//
// Try the proposer nodes last, since it's likely that they don't have a
// great view of attestations on the network.
let block = proposer_fallback
.first_success_try_proposers_last(|beacon_node| async move {
let block = match Payload::block_type() {
BlockType::Full => {
let _get_timer = metrics::start_timer_vec(
&metrics::BLOCK_SERVICE_TIMES,
&[metrics::BEACON_BLOCK_HTTP_GET],
);
beacon_node
.get_validator_blocks::<E, Payload>(
slot,
randao_reveal_ref,
graffiti.as_ref(),
)
.await
.map_err(|e| {
BlockError::Recoverable(format!(
"Error from beacon node when producing block: {:?}",
e
))
})?
.data
}
BlockType::Blinded => {
let _get_timer = metrics::start_timer_vec(
&metrics::BLOCK_SERVICE_TIMES,
&[metrics::BLINDED_BEACON_BLOCK_HTTP_GET],
);
beacon_node
.get_validator_blinded_blocks::<E, Payload>(
slot,
randao_reveal_ref,
graffiti.as_ref(),
)
.await
.map_err(|e| {
BlockError::Recoverable(format!(
"Error from beacon node when producing block: {:?}",
e
))
})?
.data
}
};
info!(
let block_contents = proposer_fallback
.first_success_try_proposers_last(|beacon_node| {
let beacon_node = beacon_node;
Self::get_validator_block(
beacon_node,
slot,
randao_reveal_ref,
graffiti,
proposer_index,
log,
"Received unsigned block";
"slot" => slot.as_u64(),
);
if proposer_index != Some(block.proposer_index()) {
return Err(BlockError::Recoverable(
"Proposer index does not match block proposer. Beacon chain re-orged"
.to_string(),
));
}
Ok::<_, BlockError>(block)
)
})
.await?;
let (block, maybe_blob_sidecars) = block_contents.deconstruct();
let signing_timer = metrics::start_timer(&metrics::BLOCK_SIGNING_TIMES);
let signed_block = match self_ref
.validator_store
.sign_block::<Payload>(*validator_pubkey_ref, block, current_slot)
@@ -541,6 +498,37 @@ impl<T: SlotClock + 'static, E: EthSpec> BlockService<T, E> {
)))
}
};
let maybe_signed_blobs = match maybe_blob_sidecars {
Some(blob_sidecars) => {
match self_ref
.validator_store
.sign_blobs::<Payload>(*validator_pubkey_ref, blob_sidecars)
.await
{
Ok(signed_blobs) => Some(signed_blobs),
Err(ValidatorStoreError::UnknownPubkey(pubkey)) => {
// A pubkey can be missing when a validator was recently removed
// via the API.
warn!(
log,
"Missing pubkey for blobs";
"info" => "a validator may have recently been removed from this VC",
"pubkey" => ?pubkey,
"slot" => ?slot
);
return Ok(());
}
Err(e) => {
return Err(BlockError::Recoverable(format!(
"Unable to sign blobs: {:?}",
e
)))
}
}
}
None => None,
};
let signing_time_ms =
Duration::from_secs_f64(signing_timer.map_or(0.0, |t| t.stop_and_record())).as_millis();
@@ -551,6 +539,8 @@ impl<T: SlotClock + 'static, E: EthSpec> BlockService<T, E> {
"signing_time_ms" => signing_time_ms,
);
let signed_block_contents = SignedBlockContents::from((signed_block, maybe_signed_blobs));
// Publish block with first available beacon node.
//
// Try the proposer nodes first, since we've likely gone to efforts to
@@ -558,39 +548,9 @@ impl<T: SlotClock + 'static, E: EthSpec> BlockService<T, E> {
// publish a block.
proposer_fallback
.first_success_try_proposers_first(|beacon_node| async {
match Payload::block_type() {
BlockType::Full => {
let _post_timer = metrics::start_timer_vec(
&metrics::BLOCK_SERVICE_TIMES,
&[metrics::BEACON_BLOCK_HTTP_POST],
);
beacon_node
.post_beacon_blocks(&signed_block)
.await
.map_err(|e| {
BlockError::Irrecoverable(format!(
"Error from beacon node when publishing block: {:?}",
e
))
})?
}
BlockType::Blinded => {
let _post_timer = metrics::start_timer_vec(
&metrics::BLOCK_SERVICE_TIMES,
&[metrics::BLINDED_BEACON_BLOCK_HTTP_POST],
);
beacon_node
.post_beacon_blinded_blocks(&signed_block)
.await
.map_err(|e| {
BlockError::Irrecoverable(format!(
"Error from beacon node when publishing block: {:?}",
e
))
})?
}
}
Ok::<_, BlockError>(())
let beacon_node = beacon_node;
self.publish_signed_block_contents::<Payload>(&signed_block_contents, beacon_node)
.await
})
.await?;
@@ -598,12 +558,132 @@ impl<T: SlotClock + 'static, E: EthSpec> BlockService<T, E> {
log,
"Successfully published block";
"block_type" => ?Payload::block_type(),
"deposits" => signed_block.message().body().deposits().len(),
"attestations" => signed_block.message().body().attestations().len(),
"deposits" => signed_block_contents.signed_block().message().body().deposits().len(),
"attestations" => signed_block_contents.signed_block().message().body().attestations().len(),
"graffiti" => ?graffiti.map(|g| g.as_utf8_lossy()),
"slot" => signed_block.slot().as_u64(),
"slot" => signed_block_contents.signed_block().slot().as_u64(),
);
Ok(())
}
async fn publish_signed_block_contents<Payload: AbstractExecPayload<E>>(
&self,
signed_block_contents: &SignedBlockContents<E, Payload>,
beacon_node: BeaconNodeHttpClient,
) -> Result<(), BlockError> {
let log = self.context.log();
let slot = signed_block_contents.signed_block().slot();
match Payload::block_type() {
BlockType::Full => {
let _post_timer = metrics::start_timer_vec(
&metrics::BLOCK_SERVICE_TIMES,
&[metrics::BEACON_BLOCK_HTTP_POST],
);
beacon_node
.clone()
.post_beacon_blocks(signed_block_contents)
.await
.or_else(|e| handle_block_post_error(e, slot, log))?
}
BlockType::Blinded => {
let _post_timer = metrics::start_timer_vec(
&metrics::BLOCK_SERVICE_TIMES,
&[metrics::BLINDED_BEACON_BLOCK_HTTP_POST],
);
beacon_node
.post_beacon_blinded_blocks(signed_block_contents)
.await
.or_else(|e| handle_block_post_error(e, slot, log))?
}
}
Ok::<_, BlockError>(())
}
async fn get_validator_block<Payload: AbstractExecPayload<E>>(
beacon_node: BeaconNodeHttpClient,
slot: Slot,
randao_reveal_ref: &SignatureBytes,
graffiti: Option<Graffiti>,
proposer_index: Option<u64>,
log: &Logger,
) -> Result<BlockContents<E, Payload>, BlockError> {
let block_contents: BlockContents<E, Payload> = match Payload::block_type() {
BlockType::Full => {
let _get_timer = metrics::start_timer_vec(
&metrics::BLOCK_SERVICE_TIMES,
&[metrics::BEACON_BLOCK_HTTP_GET],
);
beacon_node
.get_validator_blocks::<E, Payload>(slot, randao_reveal_ref, graffiti.as_ref())
.await
.map_err(|e| {
BlockError::Recoverable(format!(
"Error from beacon node when producing block: {:?}",
e
))
})?
.data
}
BlockType::Blinded => {
let _get_timer = metrics::start_timer_vec(
&metrics::BLOCK_SERVICE_TIMES,
&[metrics::BLINDED_BEACON_BLOCK_HTTP_GET],
);
beacon_node
.get_validator_blinded_blocks::<E, Payload>(
slot,
randao_reveal_ref,
graffiti.as_ref(),
)
.await
.map_err(|e| {
BlockError::Recoverable(format!(
"Error from beacon node when producing block: {:?}",
e
))
})?
.data
}
};
info!(
log,
"Received unsigned block";
"slot" => slot.as_u64(),
);
if proposer_index != Some(block_contents.block().proposer_index()) {
return Err(BlockError::Recoverable(
"Proposer index does not match block proposer. Beacon chain re-orged".to_string(),
));
}
Ok::<_, BlockError>(block_contents)
}
}
fn handle_block_post_error(err: eth2::Error, slot: Slot, log: &Logger) -> Result<(), BlockError> {
// Handle non-200 success codes.
if let Some(status) = err.status() {
if status == StatusCode::ACCEPTED {
info!(
log,
"Block is already known to BN or might be invalid";
"slot" => slot,
"status_code" => status.as_u16(),
);
return Ok(());
} else if status.is_success() {
debug!(
log,
"Block published with non-standard success code";
"slot" => slot,
"status_code" => status.as_u16(),
);
return Ok(());
}
}
Err(BlockError::Irrecoverable(format!(
"Error from beacon node when publishing block: {err:?}",
)))
}

16
validator_client/src/cli.rs
View File

@@ -170,6 +170,7 @@ pub fn cli_app<'a, 'b>() -> App<'a, 'b> {
.arg(
Arg::with_name("http-address")
.long("http-address")
.requires("http")
.value_name("ADDRESS")
.help("Set the address for the HTTP address. The HTTP server is not encrypted \
and therefore it is unsafe to publish on a public network. When this \
@@ -189,14 +190,16 @@ pub fn cli_app<'a, 'b>() -> App<'a, 'b> {
.arg(
Arg::with_name("http-port")
.long("http-port")
.requires("http")
.value_name("PORT")
.help("Set the listen TCP port for the RESTful HTTP API server.")
.default_value("5062")
.default_value_if("http", None, "5062")
.takes_value(true),
)
.arg(
Arg::with_name("http-allow-origin")
.long("http-allow-origin")
.requires("http")
.value_name("ORIGIN")
.help("Set the value of the Access-Control-Allow-Origin response HTTP header. \
Use * to allow any origin (not recommended in production). \
@@ -207,21 +210,21 @@ pub fn cli_app<'a, 'b>() -> App<'a, 'b> {
.arg(
Arg::with_name("http-allow-keystore-export")
.long("http-allow-keystore-export")
.requires("http")
.help("If present, allow access to the DELETE /lighthouse/keystores HTTP \
API method, which allows exporting keystores and passwords to HTTP API \
consumers who have access to the API token. This method is useful for \
exporting validators, however it should be used with caution since it \
exposes private key data to authorized users.")
.required(false)
.takes_value(false),
)
.arg(
Arg::with_name("http-store-passwords-in-secrets-dir")
.long("http-store-passwords-in-secrets-dir")
.requires("http")
.help("If present, any validators created via the HTTP will have keystore \
passwords stored in the secrets-dir rather than the validator \
definitions file.")
.required(false)
.takes_value(false),
)
/* Prometheus metrics HTTP server related arguments */
@@ -234,22 +237,25 @@ pub fn cli_app<'a, 'b>() -> App<'a, 'b> {
.arg(
Arg::with_name("metrics-address")
.long("metrics-address")
.requires("metrics")
.value_name("ADDRESS")
.help("Set the listen address for the Prometheus metrics HTTP server.")
.default_value("127.0.0.1")
.default_value_if("metrics", None, "127.0.0.1")
.takes_value(true),
)
.arg(
Arg::with_name("metrics-port")
.long("metrics-port")
.requires("metrics")
.value_name("PORT")
.help("Set the listen TCP port for the Prometheus metrics HTTP server.")
.default_value("5064")
.default_value_if("metrics", None, "5064")
.takes_value(true),
)
.arg(
Arg::with_name("metrics-allow-origin")
.long("metrics-allow-origin")
.requires("metrics")
.value_name("ORIGIN")
.help("Set the value of the Access-Control-Allow-Origin response HTTP header. \
Use * to allow any origin (not recommended in production). \

2
validator_client/src/config.rs
View File

@@ -8,7 +8,7 @@ use directory::{
};
use eth2::types::Graffiti;
use sensitive_url::SensitiveUrl;
use serde_derive::{Deserialize, Serialize};
use serde::{Deserialize, Serialize};
use slog::{info, warn, Logger};
use std::fs;
use std::net::IpAddr;

4
validator_client/src/doppelganger_service.rs
View File

@@ -516,9 +516,7 @@ impl DoppelgangerService {
}
// Resolve the index from the server response back to a public key.
let pubkey = if let Some(pubkey) = indices_map.get(&response.index) {
pubkey
} else {
let Some(pubkey) = indices_map.get(&response.index) else {
crit!(
self.log,
"Inconsistent indices map";

192
validator_client/src/duties_service.rs
View File

@@ -21,11 +21,12 @@ use eth2::types::{
};
use futures::{stream, StreamExt};
use parking_lot::RwLock;
use safe_arith::ArithError;
use safe_arith::{ArithError, SafeArith};
use slog::{debug, error, info, warn, Logger};
use slot_clock::SlotClock;
use std::cmp::min;
use std::collections::{hash_map, BTreeMap, HashMap, HashSet};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;
use sync::poll_sync_committee_duties;
@@ -33,14 +34,6 @@ use sync::SyncDutiesMap;
use tokio::{sync::mpsc::Sender, time::sleep};
use types::{ChainSpec, Epoch, EthSpec, Hash256, PublicKeyBytes, SelectionProof, Slot};
/// Since the BN does not like it when we subscribe to slots that are close to the current time, we
/// will only subscribe to slots which are further than `SUBSCRIPTION_BUFFER_SLOTS` away.
///
/// This number is based upon `MIN_PEER_DISCOVERY_SLOT_LOOK_AHEAD` value in the
/// `beacon_node::network::attestation_service` crate. It is not imported directly to avoid
/// bringing in the entire crate.
const SUBSCRIPTION_BUFFER_SLOTS: u64 = 2;
/// Only retain `HISTORICAL_DUTIES_EPOCHS` duties prior to the current epoch.
const HISTORICAL_DUTIES_EPOCHS: u64 = 2;
@@ -62,6 +55,36 @@ const VALIDATOR_METRICS_MIN_COUNT: usize = 64;
/// reduces the amount of data that needs to be transferred.
const INITIAL_DUTIES_QUERY_SIZE: usize = 1;
/// Offsets from the attestation duty slot at which a subscription should be sent.
const ATTESTATION_SUBSCRIPTION_OFFSETS: [u64; 8] = [3, 4, 5, 6, 7, 8, 16, 32];
/// Check that `ATTESTATION_SUBSCRIPTION_OFFSETS` is sorted ascendingly.
const _: () = assert!({
let mut i = 0;
loop {
let prev = if i > 0 {
ATTESTATION_SUBSCRIPTION_OFFSETS[i - 1]
} else {
0
};
let curr = ATTESTATION_SUBSCRIPTION_OFFSETS[i];
if curr < prev {
break false;
}
i += 1;
if i == ATTESTATION_SUBSCRIPTION_OFFSETS.len() {
break true;
}
}
});
/// Since the BN does not like it when we subscribe to slots that are close to the current time, we
/// will only subscribe to slots which are further than 2 slots away.
///
/// This number is based upon `MIN_PEER_DISCOVERY_SLOT_LOOK_AHEAD` value in the
/// `beacon_node::network::attestation_service` crate. It is not imported directly to avoid
/// bringing in the entire crate.
const _: () = assert!(ATTESTATION_SUBSCRIPTION_OFFSETS[0] > 2);
#[derive(Debug)]
pub enum Error {
UnableToReadSlotClock,
@@ -84,6 +107,16 @@ pub struct DutyAndProof {
pub duty: AttesterData,
/// This value is only set to `Some` if the proof indicates that the validator is an aggregator.
pub selection_proof: Option<SelectionProof>,
/// Track which slots we should send subscriptions at for this duty.
///
/// This value is updated after each subscription is successfully sent.
pub subscription_slots: Arc<SubscriptionSlots>,
}
/// Tracker containing the slots at which an attestation subscription should be sent.
pub struct SubscriptionSlots {
/// Pairs of `(slot, already_sent)` in slot-descending order.
slots: Vec<(Slot, AtomicBool)>,
}
impl DutyAndProof {
@@ -111,17 +144,55 @@ impl DutyAndProof {
}
})?;
let subscription_slots = SubscriptionSlots::new(duty.slot);
Ok(Self {
duty,
selection_proof,
subscription_slots,
})
}
/// Create a new `DutyAndProof` with the selection proof waiting to be filled in.
pub fn new_without_selection_proof(duty: AttesterData) -> Self {
let subscription_slots = SubscriptionSlots::new(duty.slot);
Self {
duty,
selection_proof: None,
subscription_slots,
}
}
}
impl SubscriptionSlots {
fn new(duty_slot: Slot) -> Arc<Self> {
let slots = ATTESTATION_SUBSCRIPTION_OFFSETS
.into_iter()
.filter_map(|offset| duty_slot.safe_sub(offset).ok())
.map(|scheduled_slot| (scheduled_slot, AtomicBool::new(false)))
.collect();
Arc::new(Self { slots })
}
/// Return `true` if we should send a subscription at `slot`.
fn should_send_subscription_at(&self, slot: Slot) -> bool {
// Iterate slots from smallest to largest looking for one that hasn't been completed yet.
self.slots
.iter()
.rev()
.any(|(scheduled_slot, already_sent)| {
slot >= *scheduled_slot && !already_sent.load(Ordering::Relaxed)
})
}
/// Update our record of subscribed slots to account for successful subscription at `slot`.
fn record_successful_subscription_at(&self, slot: Slot) {
for (scheduled_slot, already_sent) in self.slots.iter().rev() {
if slot >= *scheduled_slot {
already_sent.store(true, Ordering::Relaxed);
} else {
break;
}
}
}
}
@@ -570,8 +641,24 @@ async fn poll_beacon_attesters<T: SlotClock + 'static, E: EthSpec>(
let subscriptions_timer =
metrics::start_timer_vec(&metrics::DUTIES_SERVICE_TIMES, &[metrics::SUBSCRIPTIONS]);
// This vector is likely to be a little oversized, but it won't reallocate.
let mut subscriptions = Vec::with_capacity(local_pubkeys.len() * 2);
// This vector is intentionally oversized by 10% so that it won't reallocate.
// Each validator has 2 attestation duties occuring in the current and next epoch, for which
// they must send `ATTESTATION_SUBSCRIPTION_OFFSETS.len()` subscriptions. These subscription
// slots are approximately evenly distributed over the two epochs, usually with a slight lag
// that balances out (some subscriptions for the current epoch were sent in the previous, and
// some subscriptions for the next next epoch will be sent in the next epoch but aren't included
// in our calculation). We cancel the factor of 2 from the formula for simplicity.
let overallocation_numerator = 110;
let overallocation_denominator = 100;
let num_expected_subscriptions = overallocation_numerator
* std::cmp::max(
1,
local_pubkeys.len() * ATTESTATION_SUBSCRIPTION_OFFSETS.len()
/ E::slots_per_epoch() as usize,
)
/ overallocation_denominator;
let mut subscriptions = Vec::with_capacity(num_expected_subscriptions);
let mut subscription_slots_to_confirm = Vec::with_capacity(num_expected_subscriptions);
// For this epoch and the next epoch, produce any beacon committee subscriptions.
//
@@ -584,10 +671,10 @@ async fn poll_beacon_attesters<T: SlotClock + 'static, E: EthSpec>(
.read()
.iter()
.filter_map(|(_, map)| map.get(epoch))
// The BN logs a warning if we try and subscribe to current or near-by slots. Give it a
// buffer.
.filter(|(_, duty_and_proof)| {
current_slot + SUBSCRIPTION_BUFFER_SLOTS < duty_and_proof.duty.slot
duty_and_proof
.subscription_slots
.should_send_subscription_at(current_slot)
})
.for_each(|(_, duty_and_proof)| {
let duty = &duty_and_proof.duty;
@@ -599,7 +686,8 @@ async fn poll_beacon_attesters<T: SlotClock + 'static, E: EthSpec>(
committees_at_slot: duty.committees_at_slot,
slot: duty.slot,
is_aggregator,
})
});
subscription_slots_to_confirm.push(duty_and_proof.subscription_slots.clone());
});
}
@@ -624,6 +712,16 @@ async fn poll_beacon_attesters<T: SlotClock + 'static, E: EthSpec>(
"Failed to subscribe validators";
"error" => %e
)
} else {
// Record that subscriptions were successfully sent.
debug!(
log,
"Broadcast attestation subscriptions";
"count" => subscriptions.len(),
);
for subscription_slots in subscription_slots_to_confirm {
subscription_slots.record_successful_subscription_at(current_slot);
}
}
}
@@ -1184,3 +1282,67 @@ async fn notify_block_production_service<T: SlotClock + 'static, E: EthSpec>(
};
}
}
#[cfg(test)]
mod test {
use super::*;
#[test]
fn subscription_slots_exact() {
for duty_slot in [
Slot::new(32),
Slot::new(47),
Slot::new(99),
Slot::new(1002003),
] {
let subscription_slots = SubscriptionSlots::new(duty_slot);
// Run twice to check idempotence (subscription slots shouldn't be marked as done until
// we mark them manually).
for _ in 0..2 {
for offset in ATTESTATION_SUBSCRIPTION_OFFSETS {
assert!(subscription_slots.should_send_subscription_at(duty_slot - offset));
}
}
// Mark each slot as complete and check that all prior slots are still marked
// incomplete.
for (i, offset) in ATTESTATION_SUBSCRIPTION_OFFSETS
.into_iter()
.rev()
.enumerate()
{
subscription_slots.record_successful_subscription_at(duty_slot - offset);
for lower_offset in ATTESTATION_SUBSCRIPTION_OFFSETS
.into_iter()
.rev()
.skip(i + 1)
{
assert!(lower_offset < offset);
assert!(
subscription_slots.should_send_subscription_at(duty_slot - lower_offset)
);
}
}
}
}
#[test]
fn subscription_slots_mark_multiple() {
for (i, offset) in ATTESTATION_SUBSCRIPTION_OFFSETS.into_iter().enumerate() {
let duty_slot = Slot::new(64);
let subscription_slots = SubscriptionSlots::new(duty_slot);
subscription_slots.record_successful_subscription_at(duty_slot - offset);
// All past offsets (earlier slots) should be marked as complete.
for (j, other_offset) in ATTESTATION_SUBSCRIPTION_OFFSETS.into_iter().enumerate() {
let past = j >= i;
assert_eq!(other_offset >= offset, past);
assert_eq!(
subscription_slots.should_send_subscription_at(duty_slot - other_offset),
!past
);
}
}
}
}

6
validator_client/src/duties_service/sync.rs
View File

@@ -162,7 +162,7 @@ impl SyncDutiesMap {
committees_writer
.entry(committee_period)
.or_insert_with(CommitteeDuties::default)
.or_default()
.init(validator_indices);
// Return shared reference
@@ -602,9 +602,7 @@ pub async fn fill_in_aggregation_proofs<T: SlotClock + 'static, E: EthSpec>(
// Add to global storage (we add regularly so the proofs can be used ASAP).
let sync_map = duties_service.sync_duties.committees.read();
let committee_duties = if let Some(duties) = sync_map.get(&sync_committee_period) {
duties
} else {
let Some(committee_duties) = sync_map.get(&sync_committee_period) else {
debug!(
log,
"Missing sync duties";

2
validator_client/src/graffiti_file.rs
View File

@@ -1,4 +1,4 @@
use serde_derive::{Deserialize, Serialize};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::fs::File;
use std::io::{prelude::*, BufReader};

5
validator_client/src/http_api/create_signed_voluntary_exit.rs
View File

@@ -1,5 +1,6 @@
use crate::validator_store::ValidatorStore;
use bls::{PublicKey, PublicKeyBytes};
use eth2::types::GenericResponse;
use slog::{info, Logger};
use slot_clock::SlotClock;
use std::sync::Arc;
@@ -11,7 +12,7 @@ pub async fn create_signed_voluntary_exit<T: 'static + SlotClock + Clone, E: Eth
validator_store: Arc<ValidatorStore<T, E>>,
slot_clock: T,
log: Logger,
) -> Result<SignedVoluntaryExit, warp::Rejection> {
) -> Result<GenericResponse<SignedVoluntaryExit>, warp::Rejection> {
let epoch = match maybe_epoch {
Some(epoch) => epoch,
None => get_current_epoch::<T, E>(slot_clock).ok_or_else(|| {
@@ -60,7 +61,7 @@ pub async fn create_signed_voluntary_exit<T: 'static + SlotClock + Clone, E: Eth
))
})?;
Ok(signed_voluntary_exit)
Ok(GenericResponse::from(signed_voluntary_exit))
}
/// Calculates the current epoch from the genesis time and current time.

12
validator_client/src/http_api/mod.rs
View File

@@ -1214,7 +1214,8 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
.or(get_fee_recipient)
.or(get_gas_limit)
.or(get_std_keystores)
.or(get_std_remotekeys),
.or(get_std_remotekeys)
.recover(warp_utils::reject::handle_rejection),
)
.or(warp::post().and(
post_validators
@@ -1225,15 +1226,18 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
.or(post_fee_recipient)
.or(post_gas_limit)
.or(post_std_keystores)
.or(post_std_remotekeys),
.or(post_std_remotekeys)
.recover(warp_utils::reject::handle_rejection),
))
.or(warp::patch().and(patch_validators))
.or(warp::patch()
.and(patch_validators.recover(warp_utils::reject::handle_rejection)))
.or(warp::delete().and(
delete_lighthouse_keystores
.or(delete_fee_recipient)
.or(delete_gas_limit)
.or(delete_std_keystores)
.or(delete_std_remotekeys),
.or(delete_std_remotekeys)
.recover(warp_utils::reject::handle_rejection),
)),
)
// The auth route and logs are the only routes that are allowed to be accessed without the API token.

4
validator_client/src/http_api/test_utils.rs
View File

@@ -250,9 +250,9 @@ impl ApiTester {
pub async fn test_get_lighthouse_spec(self) -> Self {
let result = self
.client
.get_lighthouse_spec::<ConfigAndPresetBellatrix>()
.get_lighthouse_spec::<ConfigAndPresetCapella>()
.await
.map(|res| ConfigAndPreset::Bellatrix(res.data))
.map(|res| ConfigAndPreset::Capella(res.data))
.unwrap();
let expected = ConfigAndPreset::from_chain_spec::<E>(&E::default_spec(), None);

6
validator_client/src/http_api/tests.rs
View File

@@ -206,9 +206,9 @@ impl ApiTester {
pub async fn test_get_lighthouse_spec(self) -> Self {
let result = self
.client
.get_lighthouse_spec::<ConfigAndPresetCapella>()
.get_lighthouse_spec::<ConfigAndPresetDeneb>()
.await
.map(|res| ConfigAndPreset::Capella(res.data))
.map(|res| ConfigAndPreset::Deneb(res.data))
.unwrap();
let expected = ConfigAndPreset::from_chain_spec::<E>(&E::default_spec(), None);
@@ -503,7 +503,7 @@ impl ApiTester {
.await;
assert!(resp.is_ok());
assert_eq!(resp.unwrap().message.epoch, expected_exit_epoch);
assert_eq!(resp.unwrap().data.message.epoch, expected_exit_epoch);
self
}

4
validator_client/src/http_api/tests/keystores.rs
View File

@@ -2146,7 +2146,7 @@ async fn import_remotekey_web3signer_enabled() {
assert_eq!(tester.vals_total(), 1);
assert_eq!(tester.vals_enabled(), 1);
let vals = tester.initialized_validators.read();
let web3_vals = vals.validator_definitions().clone();
let web3_vals = vals.validator_definitions();
// Import remotekeys.
let import_res = tester
@@ -2164,7 +2164,7 @@ async fn import_remotekey_web3signer_enabled() {
assert_eq!(tester.vals_total(), 1);
assert_eq!(tester.vals_enabled(), 1);
let vals = tester.initialized_validators.read();
let remote_vals = vals.validator_definitions().clone();
let remote_vals = vals.validator_definitions();
// Web3signer should not be overwritten since it is enabled.
assert!(web3_vals == remote_vals);

5
validator_client/src/http_metrics/metrics.rs
View File

@@ -59,6 +59,11 @@ lazy_static::lazy_static! {
"Total count of attempted block signings",
&["status"]
);
pub static ref SIGNED_BLOBS_TOTAL: Result<IntCounterVec> = try_create_int_counter_vec(
"vc_signed_beacon_blobs_total",
"Total count of attempted blob signings",
&["status"]
);
pub static ref SIGNED_ATTESTATIONS_TOTAL: Result<IntCounterVec> = try_create_int_counter_vec(
"vc_signed_attestations_total",
"Total count of attempted Attestation signings",

90
validator_client/src/lib.rs
View File

@@ -82,6 +82,7 @@ const HTTP_SYNC_DUTIES_TIMEOUT_QUOTIENT: u32 = 4;
const HTTP_GET_BEACON_BLOCK_SSZ_TIMEOUT_QUOTIENT: u32 = 4;
const HTTP_GET_DEBUG_BEACON_STATE_QUOTIENT: u32 = 4;
const HTTP_GET_DEPOSIT_SNAPSHOT_QUOTIENT: u32 = 4;
const HTTP_GET_VALIDATOR_BLOCK_SSZ_TIMEOUT_QUOTIENT: u32 = 4;
const DOPPELGANGER_SERVICE_NAME: &str = "doppelganger";
@@ -98,6 +99,8 @@ pub struct ProductionValidatorClient<T: EthSpec> {
slot_clock: SystemTimeSlotClock,
http_api_listen_addr: Option<SocketAddr>,
config: Config,
beacon_nodes: Arc<BeaconNodeFallback<SystemTimeSlotClock, T>>,
genesis_time: u64,
}
impl<T: EthSpec> ProductionValidatorClient<T> {
@@ -307,6 +310,8 @@ impl<T: EthSpec> ProductionValidatorClient<T> {
/ HTTP_GET_BEACON_BLOCK_SSZ_TIMEOUT_QUOTIENT,
get_debug_beacon_states: slot_duration / HTTP_GET_DEBUG_BEACON_STATE_QUOTIENT,
get_deposit_snapshot: slot_duration / HTTP_GET_DEPOSIT_SNAPSHOT_QUOTIENT,
get_validator_block_ssz: slot_duration
/ HTTP_GET_VALIDATOR_BLOCK_SSZ_TIMEOUT_QUOTIENT,
}
} else {
Timeouts::set_all(slot_duration)
@@ -504,12 +509,6 @@ impl<T: EthSpec> ProductionValidatorClient<T> {
context.service_context("sync_committee".into()),
);
// Wait until genesis has occurred.
//
// It seems most sensible to move this into the `start_service` function, but I'm caution
// of making too many changes this close to genesis (<1 week).
wait_for_genesis(&beacon_nodes, genesis_time, &context).await?;
Ok(Self {
context,
duties_service,
@@ -522,10 +521,12 @@ impl<T: EthSpec> ProductionValidatorClient<T> {
config,
slot_clock,
http_api_listen_addr: None,
genesis_time,
beacon_nodes,
})
}
pub fn start_service(&mut self) -> Result<(), String> {
pub async fn start_service(&mut self) -> Result<(), String> {
// We use `SLOTS_PER_EPOCH` as the capacity of the block notification channel, because
// we don't expect notifications to be delayed by more than a single slot, let alone a
// whole epoch!
@@ -533,6 +534,45 @@ impl<T: EthSpec> ProductionValidatorClient<T> {
let (block_service_tx, block_service_rx) = mpsc::channel(channel_capacity);
let log = self.context.log();
let api_secret = ApiSecret::create_or_open(&self.config.validator_dir)?;
self.http_api_listen_addr = if self.config.http_api.enabled {
let ctx = Arc::new(http_api::Context {
task_executor: self.context.executor.clone(),
api_secret,
block_service: Some(self.block_service.clone()),
validator_store: Some(self.validator_store.clone()),
validator_dir: Some(self.config.validator_dir.clone()),
secrets_dir: Some(self.config.secrets_dir.clone()),
graffiti_file: self.config.graffiti_file.clone(),
graffiti_flag: self.config.graffiti,
spec: self.context.eth2_config.spec.clone(),
config: self.config.http_api.clone(),
sse_logging_components: self.context.sse_logging_components.clone(),
slot_clock: self.slot_clock.clone(),
log: log.clone(),
_phantom: PhantomData,
});
let exit = self.context.executor.exit();
let (listen_addr, server) = http_api::serve(ctx, exit)
.map_err(|e| format!("Unable to start HTTP API server: {:?}", e))?;
self.context
.clone()
.executor
.spawn_without_exit(server, "http-api");
Some(listen_addr)
} else {
info!(log, "HTTP API server is disabled");
None
};
// Wait until genesis has occurred.
wait_for_genesis(&self.beacon_nodes, self.genesis_time, &self.context).await?;
duties_service::start_update_service(self.duties_service.clone(), block_service_tx);
self.block_service
@@ -571,42 +611,6 @@ impl<T: EthSpec> ProductionValidatorClient<T> {
spawn_notifier(self).map_err(|e| format!("Failed to start notifier: {}", e))?;
let api_secret = ApiSecret::create_or_open(&self.config.validator_dir)?;
self.http_api_listen_addr = if self.config.http_api.enabled {
let ctx = Arc::new(http_api::Context {
task_executor: self.context.executor.clone(),
api_secret,
block_service: Some(self.block_service.clone()),
validator_store: Some(self.validator_store.clone()),
validator_dir: Some(self.config.validator_dir.clone()),
secrets_dir: Some(self.config.secrets_dir.clone()),
graffiti_file: self.config.graffiti_file.clone(),
graffiti_flag: self.config.graffiti,
spec: self.context.eth2_config.spec.clone(),
config: self.config.http_api.clone(),
sse_logging_components: self.context.sse_logging_components.clone(),
slot_clock: self.slot_clock.clone(),
log: log.clone(),
_phantom: PhantomData,
});
let exit = self.context.executor.exit();
let (listen_addr, server) = http_api::serve(ctx, exit)
.map_err(|e| format!("Unable to start HTTP API server: {:?}", e))?;
self.context
.clone()
.executor
.spawn_without_exit(server, "http-api");
Some(listen_addr)
} else {
info!(log, "HTTP API server is disabled");
None
};
if self.config.enable_latency_measurement_service {
latency::start_latency_service(
self.context.clone(),

6
validator_client/src/signing_method.rs
View File

@@ -37,6 +37,7 @@ pub enum Error {
pub enum SignableMessage<'a, T: EthSpec, Payload: AbstractExecPayload<T> = FullPayload<T>> {
RandaoReveal(Epoch),
BeaconBlock(&'a BeaconBlock<T, Payload>),
BlobSidecar(&'a Payload::Sidecar),
AttestationData(&'a AttestationData),
SignedAggregateAndProof(&'a AggregateAndProof<T>),
SelectionProof(Slot),
@@ -59,6 +60,7 @@ impl<'a, T: EthSpec, Payload: AbstractExecPayload<T>> SignableMessage<'a, T, Pay
match self {
SignableMessage::RandaoReveal(epoch) => epoch.signing_root(domain),
SignableMessage::BeaconBlock(b) => b.signing_root(domain),
SignableMessage::BlobSidecar(b) => b.signing_root(domain),
SignableMessage::AttestationData(a) => a.signing_root(domain),
SignableMessage::SignedAggregateAndProof(a) => a.signing_root(domain),
SignableMessage::SelectionProof(slot) => slot.signing_root(domain),
@@ -182,6 +184,10 @@ impl SigningMethod {
Web3SignerObject::RandaoReveal { epoch }
}
SignableMessage::BeaconBlock(block) => Web3SignerObject::beacon_block(block)?,
SignableMessage::BlobSidecar(_) => {
// https://github.com/ConsenSys/web3signer/issues/726
unimplemented!("Web3Signer blob signing not implemented.")
}
SignableMessage::AttestationData(a) => Web3SignerObject::Attestation(a),
SignableMessage::SignedAggregateAndProof(a) => {
Web3SignerObject::AggregateAndProof(a)

6
validator_client/src/signing_method/web3signer.rs
View File

@@ -27,6 +27,7 @@ pub enum ForkName {
Altair,
Bellatrix,
Capella,
Deneb,
}
#[derive(Debug, PartialEq, Serialize)]
@@ -95,6 +96,11 @@ impl<'a, T: EthSpec, Payload: AbstractExecPayload<T>> Web3SignerObject<'a, T, Pa
block: None,
block_header: Some(block.block_header()),
}),
BeaconBlock::Deneb(_) => Ok(Web3SignerObject::BeaconBlock {
version: ForkName::Deneb,
block: None,
block_header: Some(block.block_header()),
}),
}
}

6
validator_client/src/sync_committee_service.rs
View File

@@ -157,13 +157,11 @@ impl<T: SlotClock + 'static, E: EthSpec> SyncCommitteeService<T, E> {
.checked_sub(slot_duration / 3)
.unwrap_or_else(|| Duration::from_secs(0));
let slot_duties = if let Some(duties) = self
let Some(slot_duties) = self
.duties_service
.sync_duties
.get_duties_for_slot::<E>(slot, &self.duties_service.spec)
{
duties
} else {
else {
debug!(log, "No duties known for slot {}", slot);
return Ok(());
};

74
validator_client/src/validator_store.rs
View File

@@ -6,6 +6,7 @@ use crate::{
Config,
};
use account_utils::validator_definitions::{PasswordStorage, ValidatorDefinition};
use eth2::types::VariableList;
use parking_lot::{Mutex, RwLock};
use slashing_protection::{
interchange::Interchange, InterchangeError, NotSafe, Safe, SlashingDatabase,
@@ -17,11 +18,13 @@ use std::marker::PhantomData;
use std::path::Path;
use std::sync::Arc;
use task_executor::TaskExecutor;
use types::sidecar::Sidecar;
use types::{
attestation::Error as AttestationError, graffiti::GraffitiString, AbstractExecPayload, Address,
AggregateAndProof, Attestation, BeaconBlock, BlindedPayload, ChainSpec, ContributionAndProof,
Domain, Epoch, EthSpec, Fork, Graffiti, Hash256, Keypair, PublicKeyBytes, SelectionProof,
Signature, SignedAggregateAndProof, SignedBeaconBlock, SignedContributionAndProof, SignedRoot,
Domain, Epoch, EthSpec, Fork, ForkName, Graffiti, Hash256, Keypair, PublicKeyBytes,
SelectionProof, SidecarList, Signature, SignedAggregateAndProof, SignedBeaconBlock,
SignedContributionAndProof, SignedRoot, SignedSidecar, SignedSidecarList,
SignedValidatorRegistrationData, SignedVoluntaryExit, Slot, SyncAggregatorSelectionData,
SyncCommitteeContribution, SyncCommitteeMessage, SyncSelectionProof, SyncSubnetId,
ValidatorRegistrationData, VoluntaryExit,
@@ -369,11 +372,35 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore<T, E> {
}
fn signing_context(&self, domain: Domain, signing_epoch: Epoch) -> SigningContext {
SigningContext {
domain,
epoch: signing_epoch,
fork: self.fork(signing_epoch),
genesis_validators_root: self.genesis_validators_root,
if domain == Domain::VoluntaryExit {
match self.spec.fork_name_at_epoch(signing_epoch) {
ForkName::Base | ForkName::Altair | ForkName::Merge | ForkName::Capella => {
SigningContext {
domain,
epoch: signing_epoch,
fork: self.fork(signing_epoch),
genesis_validators_root: self.genesis_validators_root,
}
}
// EIP-7044
ForkName::Deneb => SigningContext {
domain,
epoch: signing_epoch,
fork: Fork {
previous_version: self.spec.capella_fork_version,
current_version: self.spec.capella_fork_version,
epoch: signing_epoch,
},
genesis_validators_root: self.genesis_validators_root,
},
}
} else {
SigningContext {
domain,
epoch: signing_epoch,
fork: self.fork(signing_epoch),
genesis_validators_root: self.genesis_validators_root,
}
}
}
@@ -540,6 +567,39 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore<T, E> {
}
}
pub async fn sign_blobs<Payload: AbstractExecPayload<E>>(
&self,
validator_pubkey: PublicKeyBytes,
blob_sidecars: SidecarList<E, Payload::Sidecar>,
) -> Result<SignedSidecarList<E, Payload::Sidecar>, Error> {
let mut signed_blob_sidecars = Vec::new();
for blob_sidecar in blob_sidecars.into_iter() {
let slot = blob_sidecar.slot();
let signing_epoch = slot.epoch(E::slots_per_epoch());
let signing_context = self.signing_context(Domain::BlobSidecar, signing_epoch);
let signing_method = self.doppelganger_checked_signing_method(validator_pubkey)?;
let signature = signing_method
.get_signature::<E, Payload>(
SignableMessage::BlobSidecar(blob_sidecar.as_ref()),
signing_context,
&self.spec,
&self.task_executor,
)
.await?;
metrics::inc_counter_vec(&metrics::SIGNED_BLOBS_TOTAL, &[metrics::SUCCESS]);
signed_blob_sidecars.push(SignedSidecar {
message: blob_sidecar,
signature,
_phantom: PhantomData,
});
}
Ok(VariableList::from(signed_blob_sidecars))
}
pub async fn sign_attestation(
&self,
validator_pubkey: PublicKeyBytes,