Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions crates/app/src/node/behaviour.rs
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ use libp2p::{relay, swarm::NetworkBehaviour};
use pluto_consensus::qbft;
use pluto_core::{gater::DutyGaterFn, types::PubKey};
use pluto_crypto::types::PublicKey;
use pluto_eth2api::EthBeaconNodeApiClient;
use pluto_eth2api::BeaconNodeClient;
use pluto_p2p::{
bootnode,
force_direct::ForceDirectBehaviour,
Expand Down Expand Up @@ -80,7 +80,7 @@ pub(crate) async fn wire_p2p(
peers: Vec<Peer>,
consensus: Arc<qbft::Consensus>,
duty_gater: DutyGaterFn,
eth2_cl: EthBeaconNodeApiClient,
eth2_cl: BeaconNodeClient,
pub_shares_by_key: HashMap<PubKey, HashMap<u64, PublicKey>>,
lock_hash: Vec<u8>,
builder_enabled: bool,
Expand Down
57 changes: 34 additions & 23 deletions crates/app/src/node/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -313,6 +313,11 @@ async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> {
};
let eth2_cl = build_api_client(&beacon_node_addr, config.beacon_node_timeout)?;

let beacon_client = pluto_eth2api::BeaconNodeClient::new(eth2_cl.clone());
// Broadcasting uses a separate client with the (distinct) submit timeout.
let submission_api = build_api_client(&beacon_node_addr, config.beacon_node_submit_timeout)?;
let submission_client = pluto_eth2api::BeaconNodeClient::new(submission_api);

// Fail fast if the beacon node is on a different network than the cluster
// lock (Charon's `configureEth2Client`, app.go:1022-1053). Both eth2
// clients here target the same endpoint, so a single check suffices —
Expand All @@ -322,26 +327,27 @@ async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> {
// fork version (`build_simnet_beacon_mock`), so a network mismatch is
// impossible by construction.
if simnet_beacon_mock.is_none() {
verify_fork_schedule(&eth2_cl, &lock.fork_version).await?;
verify_fork_schedule(&beacon_client, &lock.fork_version).await?;
}

let beacon_client = pluto_eth2api::BeaconNodeClient::new(eth2_cl.clone());
// Broadcasting uses a separate client with the (distinct) submit timeout.
let submission_api = build_api_client(&beacon_node_addr, config.beacon_node_submit_timeout)?;
let submission_client = pluto_eth2api::BeaconNodeClient::new(submission_api);
// Warm both config caches before duty scheduling so duty-path config
// reads start served from cache (refetched at most once per TTL);
// failure aborts startup. The network check above already filled
// `beacon_client`'s schedule slot.
tokio::try_join!(beacon_client.warm(), submission_client.warm())?;

// ---- Beacon-derived duty-workflow inputs ----

// Duty admission gate: validates duties against the beacon chain.
let duty_gater: DutyGaterFn = pluto_core::gater::DutyGater::new(&eth2_cl)
let duty_gater: DutyGaterFn = pluto_core::gater::DutyGater::new(&beacon_client)
.await
.map_err(AppError::Gater)?
.into_fn();

// Per-component deadline calculator, shared as an `Arc<dyn ...>` so a single
// beacon-derived instance backs every component's deadliner.
let deadline_calc: Arc<dyn pluto_core::deadline::DeadlineCalculator> = Arc::new(
pluto_core::deadline::DutyDeadlineCalculator::from_client(&eth2_cl)
pluto_core::deadline::DutyDeadlineCalculator::from_client(&beacon_client)
.await
.map_err(AppError::Deadline)?,
);
Expand All @@ -361,8 +367,8 @@ async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> {
// Use the mock's echoed slot timing, not the configured value: fuzz mode
// overrides it with a 12s default, and the validator mock's ticker must
// match the mock. `slots_per_epoch` also feeds the Electra activation slot.
let (fetched_slot_duration, slots_per_epoch) = eth2_cl.fetch_slots_config().await?;
let fork_config = eth2_cl.fetch_fork_config().await?;
let (fetched_slot_duration, slots_per_epoch) = beacon_client.slots_config().await?;
let fork_config = beacon_client.fork_config().await?;
let electra_slot = fork_config
.get(&pluto_eth2api::ConsensusVersion::Electra)
.map(|schedule| schedule.epoch)
Expand Down Expand Up @@ -429,7 +435,7 @@ async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> {
peers,
Arc::clone(&consensus),
Arc::clone(&duty_gater),
eth2_cl.clone(),
beacon_client.clone(),
pub_shares_by_key,
lock.lock_hash.clone(),
config.builder_api,
Expand All @@ -449,10 +455,9 @@ async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> {

// Aggregated-signature verifier: verifies the reconstructed group signature
// against the beacon-node signing domain.
let sigagg_verifier = pluto_core::sigagg::new_verifier(Arc::new(eth2_cl.clone()));
let sigagg_verifier = pluto_core::sigagg::new_verifier(beacon_client.clone());

// The readiness checker uses its own beacon-client clone, taken before
// `eth2_cl` is moved into the workflow inputs below.
// The readiness checker uses its own beacon-client clone.
let monitoring_beacon = eth2_cl.clone();

// Readiness observes which DV root pubkeys the validator client references
Expand Down Expand Up @@ -496,7 +501,6 @@ async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> {
threshold,
share_idx,
beacon_client,
eth2_cl,
submission_client,
validators,
consensus: Arc::clone(&consensus),
Expand Down Expand Up @@ -923,24 +927,31 @@ fn parse_execution_address(s: &str) -> Option<[u8; 20]> {
/// Fails fast when the beacon node is on a different network than the cluster
/// lock.
///
/// Mirrors Charon's `configureEth2Client` fork-schedule guard.
/// Mirrors Charon's `configureEth2Client` fork-schedule guard. Reads the
/// schedule through the client's config cache, so the startup warm-up reuses
/// this fetch.
async fn verify_fork_schedule(
eth2_cl: &pluto_eth2api::EthBeaconNodeApiClient,
client: &pluto_eth2api::BeaconNodeClient,
lock_fork_version: &[u8],
) -> Result<(), AppError> {
let versions = eth2_cl.fetch_fork_schedule_versions().await?;
let schedule = client.fork_schedule().await?;

if versions.iter().any(|v| v.as_slice() == lock_fork_version) {
if schedule
.iter()
.any(|fork| fork.version.as_slice() == lock_fork_version)
{
return Ok(());
}

// Best-effort network names for the operator-facing error.
Err(AppError::ForkScheduleMismatch {
lock_network: network_name_or_hex(lock_fork_version),
lock_fork_version: format!("0x{}", hex::encode(lock_fork_version)),
beacon_node_network: versions
// Defensive: `fork_schedule()` rejects empty schedules, so `first()`
// cannot be `None` here.
beacon_node_network: schedule
.first()
.map(|v| network_name_or_hex(v))
.map(|fork| network_name_or_hex(&fork.version))
.unwrap_or_else(|| "unknown".to_string()),
})
}
Expand Down Expand Up @@ -1119,7 +1130,7 @@ async fn build_simnet_validator_mock(

Ok(Arc::new(
ValidatorMock::builder()
.eth2_cl(vapi_client)
.eth2_cl(pluto_eth2api::BeaconNodeClient::new(vapi_client))
.sign_func(signer)
.pubkeys(pubshares)
.meta(meta)
Expand Down Expand Up @@ -1300,7 +1311,7 @@ mod tests {
.await
.expect("beacon mock");

verify_fork_schedule(mock.client(), &HOLESKY_FORK_VERSION)
verify_fork_schedule(&mock.beacon_client(), &HOLESKY_FORK_VERSION)
.await
.expect("matching fork version should pass the startup guard");
}
Expand All @@ -1315,7 +1326,7 @@ mod tests {
.await
.expect("beacon mock");

let err = verify_fork_schedule(mock.client(), &MAINNET_FORK_VERSION)
let err = verify_fork_schedule(&mock.beacon_client(), &MAINNET_FORK_VERSION)
.await
.expect_err("mismatched fork version should fail the startup guard");

Expand Down
18 changes: 8 additions & 10 deletions crates/app/src/node/wire.rs
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,7 @@ use pluto_core::{
validatorapi::{self, Component, Handler, SeenPubkeysFn},
};
use pluto_eth2api::{
BeaconNodeClient, EthBeaconNodeApiClient,
BeaconNodeClient,
spec::{bellatrix::ExecutionAddress, phase0::BLSPubKey},
valcache::{ValidatorCache, ValidatorCacheError},
};
Expand Down Expand Up @@ -109,10 +109,9 @@ pub struct WireInputs {
pub threshold: u64,
/// This node's 1-indexed share index.
pub share_idx: u64,
/// Beacon node client used for scheduling.
/// Beacon node client used for scheduling, fetching, validatorapi, and
/// signing-domain resolution.
pub beacon_client: BeaconNodeClient,
/// Beacon node API client used for fetching / dutydb / validatorapi.
pub eth2_cl: EthBeaconNodeApiClient,
/// Submission beacon node client used for broadcasting.
pub submission_client: BeaconNodeClient,
/// Per-validator data for this node.
Expand Down Expand Up @@ -267,7 +266,6 @@ pub async fn wire_core_workflow(
threshold,
share_idx,
beacon_client,
eth2_cl,
submission_client,
validators,
consensus,
Expand Down Expand Up @@ -301,7 +299,7 @@ pub async fn wire_core_workflow(
// would resolve duties against an empty (or unfiltered) set. `ValidatorCache`
// clones share state, so the per-epoch trim + refresh subscriber registered
// below refreshes every consumer at once.
let validator_cache = ValidatorCache::new(eth2_cl.clone(), eth2_pubkeys);
let validator_cache = ValidatorCache::new(beacon_client.api().clone(), eth2_pubkeys);
tokio::join!(
beacon_client.set_validator_cache(validator_cache.clone()),
submission_client.set_validator_cache(validator_cache.clone()),
Expand Down Expand Up @@ -386,7 +384,7 @@ pub async fn wire_core_workflow(

let fetcher = Arc::new(
Fetcher::builder()
.eth2_cl(eth2_cl.clone())
.eth2_cl(beacon_client.clone())
.fee_recipient(Arc::clone(&fee_recipient_fn))
.agg_sig_db(agg_sig_db_fn)
.await_att_data(await_att_data_fn)
Expand Down Expand Up @@ -623,7 +621,7 @@ pub async fn wire_core_workflow(
}

let (scheduler, scheduler_task) = sched_builder
.build(beacon_client, ct.clone())
.build(beacon_client.clone(), ct.clone())
.await
.map_err(AppError::Scheduler)?;

Expand All @@ -635,7 +633,7 @@ pub async fn wire_core_workflow(
// agg-attestation / sync-contribution / pubkey-by-attestation lookups, and
// the scheduler-backed duty-definition lookup.
let mut vapi = Component::new(
Arc::new(eth2_cl.clone()),
beacon_client.clone(),
Arc::clone(&dutydb),
share_idx,
pub_share_by_pubkey,
Expand Down Expand Up @@ -899,7 +897,7 @@ mod tests {
use super::*;
use pluto_core::types::SlotNumber;
use pluto_eth2api::{
BlindedBlock400Response, GetStateValidatorsResponseResponse,
BlindedBlock400Response, EthBeaconNodeApiClient, GetStateValidatorsResponseResponse,
GetStateValidatorsResponseResponseDatum, ValidatorResponseValidator, ValidatorStatus,
};
use wiremock::{
Expand Down
4 changes: 2 additions & 2 deletions crates/app/tests/wiring.rs
Original file line number Diff line number Diff line change
Expand Up @@ -253,7 +253,6 @@ fn wire_inputs_with(
threshold,
share_idx: 1,
beacon_client,
eth2_cl,
submission_client,
validators,
consensus,
Expand Down Expand Up @@ -720,7 +719,8 @@ async fn wiring_rejects_bad_partial_signature() {

// REAL eth2 verifier (mirrors production `run`): BeaconMock serves the
// signing domain via `/eth/v1/config/spec` + `/eth/v1/beacon/genesis`.
let verifier: VerifyFn = pluto_core::sigagg::new_verifier(Arc::new(eth2_cl.clone()));
let verifier: VerifyFn =
pluto_core::sigagg::new_verifier(BeaconNodeClient::new(eth2_cl.clone()));

const THRESHOLD: u64 = 2;
let inputs = wire_inputs_with(
Expand Down
Loading
Loading