diff --git a/bin/ethlambda/src/cli.rs b/bin/ethlambda/src/cli.rs index 4dbe9174..4101c56a 100644 --- a/bin/ethlambda/src/cli.rs +++ b/bin/ethlambda/src/cli.rs @@ -307,6 +307,24 @@ pub(crate) struct MainnetOptions { /// `eth-clients` publishes and kurtosis mounts at `/network-configs`. #[arg(long, default_value = crate::network::DEFAULT_NETWORK)] pub(crate) network: String, + + /// Base URL of a builder API endpoint, normally MEV-Boost (for example + /// `http://127.0.0.1:18550`) or a relay directly. Without it every block + /// is built locally, and `register_validator` answers 500 so a validator + /// client configured for a builder hears that it has none. + #[arg(long)] + pub(crate) builder: Option, + + /// How long block production waits for the builder's bid, in + /// milliseconds. Also sent to the builder as `X-Timeout-Ms`, so it can + /// hold its answer for as long as this node will wait. + #[arg( + long, + default_value_t = 1000, + requires = "builder", + value_parser = clap::value_parser!(u64).range(1..), + )] + pub(crate) builder_header_timeout: u64, } /// Flags only the lean chain takes. @@ -1020,6 +1038,56 @@ mod tests { ); } + #[test] + fn the_builder_is_off_by_default() { + let beacon = parse_beacon(beacon_args()); + + assert_eq!(beacon.mainnet.builder, None); + assert_eq!(beacon.mainnet.builder_header_timeout, 1000); + } + + #[test] + fn the_builder_flags_parse() { + let mut args = beacon_args(); + args.extend([ + "--builder", + "http://127.0.0.1:18550", + "--builder-header-timeout", + "750", + ]); + + let options: Options = parse_beacon(args).into(); + + match options.network { + Network::Mainnet { mainnet, .. } => { + assert_eq!(mainnet.builder.as_deref(), Some("http://127.0.0.1:18550")); + assert_eq!(mainnet.builder_header_timeout, 750); + } + other => panic!("expected the mainnet network, got {other:?}"), + } + } + + #[test] + fn a_zero_header_timeout_is_refused() { + let mut args = beacon_args(); + args.extend([ + "--builder", + "http://127.0.0.1:18550", + "--builder-header-timeout", + "0", + ]); + + try_parse_from(args).expect_err("a zero timeout must not parse"); + } + + #[test] + fn a_header_timeout_needs_a_builder() { + let mut args = beacon_args(); + args.extend(["--builder-header-timeout", "750"]); + + try_parse_from(args).expect_err("a timeout without a builder must not parse"); + } + #[test] fn both_execution_flags_together_pair_into_one_value() { let mut args = beacon_args(); diff --git a/bin/ethlambda/src/main.rs b/bin/ethlambda/src/main.rs index 6ea55328..def72e84 100644 --- a/bin/ethlambda/src/main.rs +++ b/bin/ethlambda/src/main.rs @@ -414,6 +414,35 @@ async fn run_node(options: Options) -> eyre::Result<()> { Network::Mainnet { .. } => false, }); + // A malformed URL fails here, at startup, rather than at the first + // proposal; an unreachable builder only warns, since it may come up + // after this node does. + let builder = match &network { + Network::Lean(_) => None, + Network::Mainnet { mainnet, .. } => mainnet + .builder + .as_deref() + .map(|url| { + ethlambda_rpc::BuilderClient::new( + url, + std::time::Duration::from_millis(mainnet.builder_header_timeout), + version::client_version(), + ) + }) + .transpose()?, + }; + if let Some(builder) = builder.clone() { + tokio::spawn(async move { + let _ = builder + .status() + .await + .inspect(|()| info!("Builder is up")) + .inspect_err(|err| { + warn!(%err, "Builder is not answering; proposals build locally until it does") + }); + }); + } + // Shared, runtime-readable sync status. The blockchain actor writes it each // tick (alongside the `lean_node_sync_status` metric); the RPC // `/lean/v0/node/syncing` endpoint reads it. Seeded to Idle, matching the @@ -805,6 +834,7 @@ async fn run_node(options: Options) -> eyre::Result<()> { events: rpc_events, client_version: version::engine_client_version(), identity: rpc_identity, + builder, }, local_peer_id, rpc_shutdown, diff --git a/crates/net/rpc/src/beacon/builder_client/mock_relay.rs b/crates/net/rpc/src/beacon/builder_client/mock_relay.rs index 01e50834..92c6d228 100644 --- a/crates/net/rpc/src/beacon/builder_client/mock_relay.rs +++ b/crates/net/rpc/src/beacon/builder_client/mock_relay.rs @@ -6,6 +6,7 @@ //! registration signatures and refuses SSZ registrations, records blinded //! block submissions, and has `answer_headers_with` and //! `answer_submissions_with` to answer a call with exactly what a test needs. +//! It serves `status` with 200, or 503 after `make_unavailable`. use std::sync::{Arc, Mutex}; use std::time::Duration; @@ -51,6 +52,10 @@ struct RelayState { submit_status: Option, /// Submissions that arrived SSZ-encoded, refused or not. ssz_submissions: usize, + /// Whether `status` answers 503. + unavailable: bool, + /// `registerValidator` calls received, whatever their outcome. + registration_calls: usize, } /// A fixed answer to every `getHeader`. @@ -92,6 +97,7 @@ impl MockRelay { ) .route("/eth/v1/builder/validators", post(register)) .route("/eth/v2/builder/blinded_blocks", post(submit)) + .route("/eth/v1/builder/status", get(status)) .with_state(state.clone()); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let url = format!("http://{}", listener.local_addr().unwrap()); @@ -148,6 +154,11 @@ impl MockRelay { self.state.lock().unwrap().submit_status = Some(status); } + /// Answer `status` with 503, as a builder that is not ready does. + pub(crate) fn make_unavailable(&self) { + self.state.lock().unwrap().unavailable = true; + } + pub(crate) fn blinded_blocks(&self) -> Vec { self.state.lock().unwrap().blinded_blocks.clone() } @@ -156,6 +167,11 @@ impl MockRelay { self.state.lock().unwrap().ssz_submissions } + /// How many `registerValidator` calls arrived, whatever their outcome. + pub(crate) fn registration_calls(&self) -> usize { + self.state.lock().unwrap().registration_calls + } + pub(crate) fn registrations(&self) -> Vec { self.state.lock().unwrap().registrations.clone() } @@ -244,6 +260,7 @@ async fn header( async fn register(State(state): State, headers: HeaderMap, body: Bytes) -> Response { let mut state = state.lock().unwrap(); + state.registration_calls += 1; // The client always registers in JSON (see `register_validators`); an // SSZ registration here is a regression. if is_ssz_request(&headers) { @@ -264,6 +281,14 @@ async fn register(State(state): State, headers: HeaderMap, body: Bytes) StatusCode::OK.into_response() } +async fn status(State(state): State) -> Response { + if state.lock().unwrap().unavailable { + StatusCode::SERVICE_UNAVAILABLE.into_response() + } else { + StatusCode::OK.into_response() + } +} + async fn submit(State(state): State, headers: HeaderMap, body: Bytes) -> Response { let mut state = state.lock().unwrap(); let ssz = is_ssz_request(&headers); diff --git a/crates/net/rpc/src/beacon/builder_client/mod.rs b/crates/net/rpc/src/beacon/builder_client/mod.rs index 4ebf7ca8..d702e3d0 100644 --- a/crates/net/rpc/src/beacon/builder_client/mod.rs +++ b/crates/net/rpc/src/beacon/builder_client/mod.rs @@ -14,6 +14,8 @@ //! Responses are decoded by their own `Content-Type`, not by what was asked //! for: a builder may answer JSON to an SSZ request without refusing it. //! +//! `status` is a liveness probe the node calls once at startup. +//! //! Every call is timed into `lean_beacon_builder_request_seconds`, by call //! and outcome. @@ -63,6 +65,9 @@ const REGISTER_TIMEOUT: Duration = Duration::from_secs(10); /// fetching the payload from its relays before it answers. const SUBMIT_TIMEOUT: Duration = Duration::from_secs(5); +/// `status` is a liveness probe; a builder that takes longer is not ready. +const STATUS_TIMEOUT: Duration = Duration::from_secs(5); + /// A builder API endpoint: a relay, or MEV-Boost in front of several. #[derive(Debug, Clone)] pub struct BuilderClient { @@ -274,6 +279,30 @@ impl BuilderClient { .map_err(|err| BuilderError::from_reqwest(err, SUBMIT_TIMEOUT)) } + /// `status`: whether the builder is up. The node calls it once at + /// startup, so a wrong `--builder` shows before the first proposal + /// rather than at it. + pub async fn status(&self) -> Result<(), BuilderError> { + let started = Instant::now(); + let result = self.try_status().await; + observe_builder_request("status", outcome(&result), started.elapsed()); + result + } + + async fn try_status(&self) -> Result<(), BuilderError> { + let response = self + .http + .get(self.url("/eth/v1/builder/status")) + .timeout(STATUS_TIMEOUT) + .send() + .await + .map_err(|err| BuilderError::from_reqwest(err, STATUS_TIMEOUT))?; + match response.status() { + StatusCode::OK => Ok(()), + _ => Err(BuilderError::from_status(response, STATUS_TIMEOUT).await), + } + } + fn fall_back_to_json(&self) { info!("Builder refused SSZ; using JSON from now on"); self.json_only.store(true, Ordering::Relaxed); diff --git a/crates/net/rpc/src/beacon/builder_client/tests.rs b/crates/net/rpc/src/beacon/builder_client/tests.rs index 59691376..b2874c53 100644 --- a/crates/net/rpc/src/beacon/builder_client/tests.rs +++ b/crates/net/rpc/src/beacon/builder_client/tests.rs @@ -393,3 +393,19 @@ async fn only_202_is_a_successful_submission() { )); } } + +#[tokio::test] +async fn a_running_builder_reports_its_status() { + let relay = MockRelay::start(GENESIS_FORK_VERSION).await; + client_for(&relay).status().await.unwrap(); +} + +#[tokio::test] +async fn an_unavailable_builder_reports_its_503() { + let relay = MockRelay::start(GENESIS_FORK_VERSION).await; + relay.make_unavailable(); + assert!(matches!( + client_for(&relay).status().await, + Err(BuilderError::Status { status: 503, .. }) + )); +} diff --git a/crates/net/rpc/src/beacon/mod.rs b/crates/net/rpc/src/beacon/mod.rs index 0c707791..dc6af10c 100644 --- a/crates/net/rpc/src/beacon/mod.rs +++ b/crates/net/rpc/src/beacon/mod.rs @@ -34,6 +34,7 @@ pub(crate) mod pool; pub(crate) mod proposal; pub(crate) mod proposer_preferences; pub(crate) mod ptc; +pub(crate) mod registrations; pub(crate) mod states; pub(crate) mod sync_committee; pub(crate) mod validator; @@ -191,6 +192,7 @@ pub(crate) fn routes(version: &'static str, peer_id: String) -> Router { .merge(config::routes()) .merge(node::routes(version, peer_id)) .merge(validator::routes()) + .merge(registrations::routes()) .merge(pool::routes()) .merge(operations::routes()) .merge(proposal::routes()) diff --git a/crates/net/rpc/src/beacon/registrations.rs b/crates/net/rpc/src/beacon/registrations.rs new file mode 100644 index 00000000..bca74807 --- /dev/null +++ b/crates/net/rpc/src/beacon/registrations.rs @@ -0,0 +1,503 @@ +//! `POST /eth/v1/validator/register_validator`: a validator client's builder +//! registrations, checked against the head state and forwarded to the +//! builder `--builder` names. +//! +//! Validators that the head state does not hold, or holds as exited, are +//! dropped and counted rather than refused. Exit is judged at the later of +//! the head's epoch and the wall clock's, so a lagging head cannot let an +//! exited validator through: the beacon-APIs say they must +//! not reach the builder, not that the request fails. Each kept +//! registration also becomes its validator's fee recipient, as +//! `prepare_beacon_proposer` would make it, so a block built locally after +//! all still pays the address the validator registered. + +use std::collections::{HashMap, HashSet}; + +use axum::body::Bytes; +use axum::extract::{DefaultBodyLimit, State}; +use axum::http::HeaderMap; +use axum::response::{IntoResponse, Response}; +use axum::routing::post; +use axum::{Extension, Router}; +use ethlambda_storage::Store; +use ethlambda_types::beacon::builder::{ + SignedValidatorRegistrationV1, SignedValidatorRegistrations, +}; +use ethlambda_types::beacon::containers::BeaconState; +use ethlambda_types::beacon::primitives::{BlsPubkey, Epoch, ValidatorIndex}; +use ethlambda_types::beacon::signing::compute_epoch_at_slot; +use libssz::SszDecode as _; +use tracing::warn; + +use super::builder_client::{BuilderClient, BuilderError}; +use super::validator::{FeeRecipients, head}; +use super::{ApiError, BodyEncoding}; +use crate::metrics::inc_builder_registrations; + +pub(crate) fn routes() -> Router { + // A validator client may send every key it holds at once, which is more + // than axum's default body limit takes. + Router::new().route( + "/eth/v1/validator/register_validator", + post(post_register_validator).layer(DefaultBodyLimit::max(super::MAX_PUBLISH_BODY_BYTES)), + ) +} + +async fn post_register_validator( + State(store): State, + Extension(builder): Extension>, + Extension(fee_recipients): Extension, + headers: HeaderMap, + body: Bytes, +) -> Response { + register(store, builder, fee_recipients, &headers, &body) + .await + .map_or_else(IntoResponse::into_response, |()| { + axum::http::StatusCode::OK.into_response() + }) +} + +async fn register( + store: Store, + builder: Option, + fee_recipients: FeeRecipients, + headers: &HeaderMap, + body: &[u8], +) -> Result<(), ApiError> { + // Decoded first, so a malformed body is a 400 whether or not a builder is + // configured. + let registrations = decode(headers, body)?; + let builder = builder.ok_or(ApiError::Internal("no builder is configured (--builder)"))?; + + // The head-state read and the walk over the whole registry are blocking + // work, as the duty endpoints' walks are. + let submitted = registrations.len(); + let kept = tokio::task::spawn_blocking(move || { + let (_, state) = head(&store)?; + // Judged at the later of the head's epoch and the wall clock's, so a + // lagging head cannot let an exited validator through. + let wall_epoch = compute_epoch_at_slot(super::node::wall_slot(&store)); + Ok::<_, ApiError>(active_or_pending(&state, registrations, wall_epoch)) + }) + .await + .map_err(|_| ApiError::Internal("checking the registrations failed"))??; + inc_builder_registrations("dropped", (submitted - kept.len()) as u64); + if kept.is_empty() { + return Ok(()); + } + + { + let mut recipients = fee_recipients.lock().expect("fee recipient lock poisoned"); + for (index, registration) in &kept { + recipients.insert(*index, registration.message.fee_recipient); + } + } + let kept: Vec = kept + .into_iter() + .map(|(_, registration)| registration) + .collect(); + // On its own task: a validator client that hangs up drops this handler, + // and the forward, its count and its log must complete regardless. + let forwarded = tokio::spawn(async move { + let count = kept.len() as u64; + let result = builder.register_validators(&kept).await; + match &result { + Ok(()) => inc_builder_registrations("forwarded", count), + Err(err) => { + inc_builder_registrations("failed", count); + warn!(%err, count, "Builder did not take the registrations"); + } + } + result + }) + .await + .map_err(|_| ApiError::Internal("forwarding the registrations failed"))?; + forwarded.map_err(|err| match err { + BuilderError::BadRequest(message) => ApiError::BadRequestDetail(message), + _ => ApiError::Internal("the builder did not take the registrations"), + }) +} + +/// The request body: a JSON array, or the SSZ list builder-specs defines. +fn decode( + headers: &HeaderMap, + body: &[u8], +) -> Result, ApiError> { + match BodyEncoding::from_headers(headers)? { + BodyEncoding::Ssz => SignedValidatorRegistrations::from_ssz_bytes(body) + .map(|list| list.to_vec()) + .map_err(|_| ApiError::BadRequest("invalid SSZ registrations")), + BodyEncoding::Json => serde_json::from_slice(body) + .map_err(|_| ApiError::BadRequest("invalid JSON registrations")), + } +} + +/// The registrations whose validator `state` holds as active or pending at +/// the later of its own epoch and `wall_epoch`, each with its validator +/// index. One pass over the registry, whatever the number of registrations. +fn active_or_pending( + state: &BeaconState, + registrations: Vec, + wall_epoch: Epoch, +) -> Vec<(ValidatorIndex, SignedValidatorRegistrationV1)> { + let epoch = compute_epoch_at_slot(state.slot()).max(wall_epoch); + let wanted: HashSet = registrations + .iter() + .map(|registration| registration.message.pubkey) + .collect(); + // Active or pending is exactly "not yet exited": a pending validator's + // exit epoch is still FAR_FUTURE_EPOCH. + let indices: HashMap = state + .iter_validators() + .enumerate() + .filter(|(_, validator)| epoch < validator.exit_epoch && wanted.contains(&validator.pubkey)) + .map(|(index, validator)| (validator.pubkey, index as ValidatorIndex)) + .collect(); + registrations + .into_iter() + .filter_map(|registration| { + let index = indices.get(®istration.message.pubkey)?; + Some((*index, registration)) + }) + .collect() +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use axum::body::Body; + use axum::http::{Request, StatusCode}; + use ethlambda_state_transition::beacon::helpers::test_state::with_signing_validators_at; + use ethlambda_types::beacon::constants::FAR_FUTURE_EPOCH; + use ethlambda_types::beacon::fork::ForkName; + use ethlambda_types::beacon::primitives::{ExecutionAddress, Version}; + use http_body_util::BodyExt as _; + use libssz::SszEncode as _; + use tower::ServiceExt as _; + + use super::*; + use crate::beacon::builder_client::mock_relay::{MockRelay, signed_registration}; + use crate::test_utils::beacon_store_at; + + const VALIDATORS: usize = 8; + + /// A store whose head state holds `VALIDATORS` active validators, minus + /// the indices in `exited`, which have already left. + fn store_with_exited(exited: &[usize]) -> Store { + let mut state = with_signing_validators_at(ForkName::Fulu, VALIDATORS); + for &index in exited { + state + .validator_mut(index as ValidatorIndex) + .unwrap() + .exit_epoch = 0; + } + state.apply_pending_mutations(); + beacon_store_at(state).0 + } + + fn genesis_fork_version(store: &Store) -> Version { + store.config().genesis_fork_version + } + + async fn post( + store: Store, + builder: Option, + fee_recipients: FeeRecipients, + request: Request, + ) -> (StatusCode, String) { + let app = routes() + .with_state(store) + .layer(Extension(builder)) + .layer(Extension(fee_recipients)); + let response = app.oneshot(request).await.unwrap(); + let status = response.status(); + let body = response.into_body().collect().await.unwrap().to_bytes(); + (status, String::from_utf8_lossy(&body).into_owned()) + } + + fn json_request(registrations: &[SignedValidatorRegistrationV1]) -> Request { + Request::post("/eth/v1/validator/register_validator") + .header("content-type", "application/json") + .body(Body::from(serde_json::to_vec(registrations).unwrap())) + .unwrap() + } + + fn client_for(relay: &MockRelay) -> Option { + Some( + BuilderClient::new(relay.url.clone(), Duration::from_secs(1), "ethlambda/test") + .unwrap(), + ) + } + + #[tokio::test] + async fn known_validators_reach_the_builder_and_set_their_fee_recipients() { + let store = store_with_exited(&[]); + let relay = MockRelay::start(genesis_fork_version(&store)).await; + let registrations = vec![ + signed_registration(0, genesis_fork_version(&store)), + signed_registration(1, genesis_fork_version(&store)), + ]; + let fee_recipients = FeeRecipients::default(); + + let (status, body) = post( + store, + client_for(&relay), + fee_recipients.clone(), + json_request(®istrations), + ) + .await; + + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(relay.registrations(), registrations); + let recipients = fee_recipients.lock().unwrap(); + assert_eq!( + recipients.get(&0), + Some(&ExecutionAddress::repeat_byte(0x11)) + ); + assert_eq!( + recipients.get(&1), + Some(&ExecutionAddress::repeat_byte(0x11)) + ); + } + + #[tokio::test] + async fn unknown_and_exited_validators_are_dropped() { + let store = store_with_exited(&[2]); + let relay = MockRelay::start(genesis_fork_version(&store)).await; + let kept = signed_registration(0, genesis_fork_version(&store)); + let exited = signed_registration(2, genesis_fork_version(&store)); + let unknown = signed_registration(VALIDATORS + 10, genesis_fork_version(&store)); + let fee_recipients = FeeRecipients::default(); + + let (status, body) = post( + store, + client_for(&relay), + fee_recipients.clone(), + json_request(&[kept.clone(), exited, unknown]), + ) + .await; + + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(relay.registrations(), [kept]); + assert_eq!(fee_recipients.lock().unwrap().len(), 1); + } + + #[tokio::test] + async fn nothing_left_to_forward_is_a_success_without_a_call() { + let store = store_with_exited(&[]); + let relay = MockRelay::start(genesis_fork_version(&store)).await; + let unknown = signed_registration(VALIDATORS + 10, genesis_fork_version(&store)); + + let (status, body) = post( + store, + client_for(&relay), + FeeRecipients::default(), + json_request(&[unknown]), + ) + .await; + + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(relay.registration_calls(), 0); + } + + #[tokio::test] + async fn without_a_builder_the_request_is_a_server_error() { + let store = store_with_exited(&[]); + let registration = signed_registration(0, genesis_fork_version(&store)); + let (status, _) = post( + store, + None, + FeeRecipients::default(), + json_request(&[registration]), + ) + .await; + assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR); + } + + #[tokio::test] + async fn a_builders_refusal_is_passed_through_as_a_400() { + let store = store_with_exited(&[]); + let relay = MockRelay::start(genesis_fork_version(&store)).await; + let mut forged = signed_registration(0, genesis_fork_version(&store)); + forged.signature = signed_registration(1, genesis_fork_version(&store)).signature; + + let (status, body) = post( + store, + client_for(&relay), + FeeRecipients::default(), + json_request(&[forged]), + ) + .await; + + assert_eq!(status, StatusCode::BAD_REQUEST); + assert!(body.contains("invalid signature"), "{body}"); + } + + #[tokio::test] + async fn an_unreachable_builder_is_a_server_error() { + let store = store_with_exited(&[]); + let registration = signed_registration(0, genesis_fork_version(&store)); + let builder = BuilderClient::new( + "http://127.0.0.1:1", + Duration::from_secs(1), + "ethlambda/test", + ) + .unwrap(); + let (status, _) = post( + store, + Some(builder), + FeeRecipients::default(), + json_request(&[registration]), + ) + .await; + assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR); + } + + #[tokio::test] + async fn an_ssz_body_is_accepted() { + let store = store_with_exited(&[]); + let relay = MockRelay::start(genesis_fork_version(&store)).await; + let registrations = vec![signed_registration(3, genesis_fork_version(&store))]; + let list = SignedValidatorRegistrations::try_from(registrations.clone()).unwrap(); + let request = Request::post("/eth/v1/validator/register_validator") + .header("content-type", "application/octet-stream") + .body(Body::from(list.to_ssz())) + .unwrap(); + + let (status, body) = + post(store, client_for(&relay), FeeRecipients::default(), request).await; + + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(relay.registrations(), registrations); + } + + /// A store whose validator `index` is pending: its activation is in the + /// future and its exit is `FAR_FUTURE_EPOCH`. + fn store_with_pending(index: usize) -> Store { + let mut state = with_signing_validators_at(ForkName::Fulu, VALIDATORS); + let validator = state.validator_mut(index as ValidatorIndex).unwrap(); + validator.activation_epoch = 1_000; + validator.exit_epoch = FAR_FUTURE_EPOCH; + state.apply_pending_mutations(); + beacon_store_at(state).0 + } + + #[tokio::test] + async fn a_pending_validator_is_kept() { + let store = store_with_pending(4); + let relay = MockRelay::start(genesis_fork_version(&store)).await; + let pending = signed_registration(4, genesis_fork_version(&store)); + + let (status, body) = post( + store, + client_for(&relay), + FeeRecipients::default(), + json_request(std::slice::from_ref(&pending)), + ) + .await; + + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(relay.registrations(), [pending]); + } + + #[tokio::test] + async fn fee_recipients_are_set_even_when_the_builder_refuses() { + let store = store_with_exited(&[]); + let relay = MockRelay::start(genesis_fork_version(&store)).await; + let mut forged = signed_registration(0, genesis_fork_version(&store)); + forged.signature = signed_registration(1, genesis_fork_version(&store)).signature; + let fee_recipients = FeeRecipients::default(); + + let (status, _) = post( + store, + client_for(&relay), + fee_recipients.clone(), + json_request(&[forged]), + ) + .await; + + assert_eq!(status, StatusCode::BAD_REQUEST); + assert_eq!( + fee_recipients.lock().unwrap().get(&0), + Some(&ExecutionAddress::repeat_byte(0x11)) + ); + } + + #[tokio::test] + async fn fee_recipients_are_set_even_when_the_builder_is_unreachable() { + let store = store_with_exited(&[]); + let registration = signed_registration(0, genesis_fork_version(&store)); + let builder = BuilderClient::new( + "http://127.0.0.1:1", + Duration::from_secs(1), + "ethlambda/test", + ) + .unwrap(); + let fee_recipients = FeeRecipients::default(); + + let (status, _) = post( + store, + Some(builder), + fee_recipients.clone(), + json_request(&[registration]), + ) + .await; + + assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR); + assert_eq!( + fee_recipients.lock().unwrap().get(&0), + Some(&ExecutionAddress::repeat_byte(0x11)) + ); + } + + #[tokio::test] + async fn a_malformed_body_is_a_400() { + let store = store_with_exited(&[]); + let request = Request::post("/eth/v1/validator/register_validator") + .header("content-type", "application/json") + .body(Body::from("not json")) + .unwrap(); + + let (status, body) = post(store, None, FeeRecipients::default(), request).await; + + assert_eq!(status, StatusCode::BAD_REQUEST, "{body}"); + } + + #[tokio::test] + async fn a_content_type_other_than_json_or_ssz_is_a_415() { + let store = store_with_exited(&[]); + let request = Request::post("/eth/v1/validator/register_validator") + .header("content-type", "text/plain") + .body(Body::from("[]")) + .unwrap(); + + let (status, body) = post(store, None, FeeRecipients::default(), request).await; + + assert_eq!(status, StatusCode::UNSUPPORTED_MEDIA_TYPE, "{body}"); + } + + #[tokio::test] + async fn an_exit_the_wall_clock_has_passed_drops_the_validator() { + // The head is at epoch 0 and the clock at epoch 10: an exit at epoch + // 5 is in the future for the head and in the past for the clock. + let mut state = with_signing_validators_at(ForkName::Fulu, VALIDATORS); + state.validator_mut(2).unwrap().exit_epoch = 5; + state.apply_pending_mutations(); + let (store, _) = crate::test_utils::beacon_store_at_clock(state, 32 * 10); + let relay = MockRelay::start(genesis_fork_version(&store)).await; + let kept = signed_registration(0, genesis_fork_version(&store)); + let exited = signed_registration(2, genesis_fork_version(&store)); + + let (status, body) = post( + store, + client_for(&relay), + FeeRecipients::default(), + json_request(&[kept.clone(), exited]), + ) + .await; + + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(relay.registrations(), [kept]); + } +} diff --git a/crates/net/rpc/src/lib.rs b/crates/net/rpc/src/lib.rs index 11ddca14..5098ef54 100644 --- a/crates/net/rpc/src/lib.rs +++ b/crates/net/rpc/src/lib.rs @@ -238,6 +238,9 @@ pub struct BeaconApiHandles { pub client_version: ethlambda_engine::types::ClientVersionV1, /// What `/eth/v1/node/identity` reports beyond the peer id. pub identity: BeaconIdentity, + /// The builder `--builder` names, for the builder flow's endpoints. + /// `None` when the node runs without one. + pub builder: Option, } /// Start the HTTP servers for a beacon node. @@ -263,6 +266,7 @@ pub async fn start_beacon_rpc_server( .layer(Extension(handles.custody_columns)) .layer(Extension(beacon::validator::FeeRecipients::default())) .layer(Extension(handles.engine)) + .layer(Extension(handles.builder)) .layer(Extension(handles.events)) .layer(Extension(handles.identity)) .layer(Extension(beacon::graffiti::OwnVersion(Arc::new( diff --git a/crates/net/rpc/src/metrics.rs b/crates/net/rpc/src/metrics.rs index bb1529e6..14baad83 100644 --- a/crates/net/rpc/src/metrics.rs +++ b/crates/net/rpc/src/metrics.rs @@ -1,6 +1,7 @@ use axum::{Router, http::HeaderValue, http::header, response::IntoResponse, routing::get}; use ethlambda_metrics::{ - Histogram, HistogramVec, gather_default_metrics, register_histogram, register_histogram_vec, + Histogram, HistogramVec, IntCounterVec, gather_default_metrics, register_histogram, + register_histogram_vec, register_int_counter_vec, }; use std::{sync::LazyLock, time::Duration}; use tracing::warn; @@ -66,3 +67,20 @@ pub(crate) async fn get_metrics() -> impl IntoResponse { .insert(header::CONTENT_TYPE, content_type); response } + +/// Count `register_validator` entries by what became of them: forwarded to +/// the builder, failed at it, or dropped here as an unknown or exited +/// validator. +pub(crate) fn inc_builder_registrations(outcome: &str, count: u64) { + static LEAN_BEACON_BUILDER_REGISTRATIONS_TOTAL: LazyLock = LazyLock::new(|| { + register_int_counter_vec!( + "lean_beacon_builder_registrations_total", + "Validator registrations received, by what became of them", + &["outcome"] + ) + .unwrap() + }); + LEAN_BEACON_BUILDER_REGISTRATIONS_TOTAL + .with_label_values(&[outcome]) + .inc_by(count); +} diff --git a/docs/cli.md b/docs/cli.md index 2dcbf670..489b24f9 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -114,6 +114,8 @@ from a normal build. | `--execution-endpoint` | none | Base URL of the execution client's Engine API endpoint, e.g. `http://127.0.0.1:8551`. Must be given together with `--execution-jwt-secret` | | `--execution-jwt-secret` | none | File holding the 32-byte hex JWT secret shared with the execution client | | `--safe-slots-to-import-optimistically` | the specification's own value | How far behind the wall clock a block must be before it may be imported optimistically on age alone | +| `--builder` | none | Base URL of a builder API endpoint, normally MEV-Boost (e.g. `http://127.0.0.1:18550`). Without it every block is built locally | +| `--builder-header-timeout` | `1000` | Milliseconds block production waits for the builder's bid; also sent to the builder as `X-Timeout-Ms`. Requires `--builder` | Accepted range is `CUSTODY_REQUIREMENT` to `NUMBER_OF_CUSTODY_GROUPS` (4 to 128), enforced at parse time rather than clamped: serving a different set than diff --git a/docs/metrics.md b/docs/metrics.md index 48616f7b..5e3edce4 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -467,7 +467,8 @@ ethlambda-specific, not part of the leanMetrics spec. | Name | Type | Usage | Sample collection event | Labels | Buckets | |------|------|-------|-------------------------|--------|---------| | `lean_beacon_publish_data_columns_seconds` | Histogram | Time to compute and verify a published block's cells and build its data column sidecars | On every published block that carries blobs and whose cells verify | | 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5 | -| `lean_beacon_builder_request_seconds` | Histogram | Time a builder API call took | On every `getHeader`, `registerValidator` and `submitBlindedBlockV2` call to the configured builder | call=get_header,register_validators,submit_blinded_block
outcome=ok,no_bid,timeout,unreachable,bad_request,unexpected_status,invalid_response,unsupported_fork,encode | 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 0.75, 1, 2, 5, 10 | +| `lean_beacon_builder_request_seconds` | Histogram | Time a builder API call took | On every `getHeader`, `registerValidator`, `submitBlindedBlockV2` and `status` call to the configured builder | call=get_header,register_validators,submit_blinded_block,status
outcome=ok,no_bid,timeout,unreachable,bad_request,unexpected_status,invalid_response,unsupported_fork,encode | 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 0.75, 1, 2, 5, 10 | +| `lean_beacon_builder_registrations_total` | Counter | Validator registrations received, by what became of them | On every `register_validator` request | outcome=forwarded,failed,dropped | | ### Beacon Committee Cache diff --git a/docs/rpc.md b/docs/rpc.md index 04d72589..85e8aaea 100644 --- a/docs/rpc.md +++ b/docs/rpc.md @@ -274,6 +274,7 @@ surface rather than sitting beside it; a `/lean/v0` path on a beacon node is a | `POST` | `/eth/v1/validator/proposer_preferences` | *(status only)* | Validate, cache and gossip `SignedProposerPreferences` (gloas; JSON or SSZ body) | | `POST` | `/eth/v1/beacon/states/{state_id}/builders` | JSON | The gloas builder registry, filtered by id and status | | `POST` | `/eth/v1/validator/prepare_beacon_proposer` | *(status only)* | Record each validator's fee recipient for block production (see below) | +| `POST` | `/eth/v1/validator/register_validator` | *(status only)* | Forwards builder registrations of active or pending validators to `--builder`, and sets their fee recipients | | `GET` | `/eth/v1/events` | SSE | Live stream of chain events (see below) | ### Validator endpoints @@ -409,6 +410,17 @@ the chain actor writes, so no request waits on the actor. once a slot. - **`prepare_beacon_proposer`** records each validator's fee recipient, in memory (a validator client repeats the call every epoch). +- **`register_validator`** takes a JSON array or an SSZ list of signed + builder registrations. Registrations of validators the head state does not + hold, or holds as exited, are dropped and counted + (`lean_beacon_builder_registrations_total`), not refused. Exit is judged at + the later of the head's epoch and the wall clock's, so a lagging head cannot + let an exited validator through. The beacon-APIs say + they must not reach the builder, not that the request fails. Each kept + registration also sets its validator's fee recipient, so a block built + locally still pays the registered address. The kept ones go to `--builder` in + one call, skipped when none are left. A builder's 400 is passed through and + any other failure is a 500; with no `--builder` the answer is 500. - **`blocks/{slot}`** advances the head state to the slot and asks the node's own execution client to build on the head (`forkchoiceUpdatedV3` with payload attributes, then `getPayloadV5`), with the proposer's fee recipient.