From c6dba21c918d4ed8b6ba5d4b7f7384336672f092 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Mon, 26 Oct 2026 14:05:19 -0300 Subject: [PATCH 1/6] feat(rpc): ask a builder whether it is up builder-specs' status endpoint, so the node can check its builder once at startup and a wrong --builder shows before the first proposal rather than as a missing bid at it. --- .../src/beacon/builder_client/mock_relay.rs | 17 +++++++++++ .../net/rpc/src/beacon/builder_client/mod.rs | 29 +++++++++++++++++++ .../rpc/src/beacon/builder_client/tests.rs | 16 ++++++++++ docs/metrics.md | 2 +- 4 files changed, 63 insertions(+), 1 deletion(-) 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..1e0285c0 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,8 @@ struct RelayState { submit_status: Option, /// Submissions that arrived SSZ-encoded, refused or not. ssz_submissions: usize, + /// Whether `status` answers 503. + unavailable: bool, } /// A fixed answer to every `getHeader`. @@ -92,6 +95,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 +152,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() } @@ -264,6 +273,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/docs/metrics.md b/docs/metrics.md index 48616f7b..01ac03ab 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -467,7 +467,7 @@ 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 | ### Beacon Committee Cache From 0d5a9cdc72b0ad79ef7e3c62776e73905d5d9621 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Mon, 26 Oct 2026 14:06:28 -0300 Subject: [PATCH 2/6] feat(beacon): point the node at a builder with --builder The client is built at startup, so a malformed URL stops the node there, and its status is checked once in the background, warning when the builder is not up yet rather than failing a node that would only lose MEV until it is. --builder-header-timeout bounds the wait for a bid and is sent to the builder as X-Timeout-Ms. --- bin/ethlambda/src/cli.rs | 50 +++++++++++++++++++++++++++++++++++++++ bin/ethlambda/src/main.rs | 30 +++++++++++++++++++++++ crates/net/rpc/src/lib.rs | 4 ++++ docs/cli.md | 2 ++ 4 files changed, 86 insertions(+) diff --git a/bin/ethlambda/src/cli.rs b/bin/ethlambda/src/cli.rs index 4dbe9174..d82daf0c 100644 --- a/bin/ethlambda/src/cli.rs +++ b/bin/ethlambda/src/cli.rs @@ -307,6 +307,19 @@ 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")] + pub(crate) builder_header_timeout: u64, } /// Flags only the lean chain takes. @@ -1020,6 +1033,43 @@ 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_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/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/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 From 911f226c4fc30999513afff45a0e4748684df4c3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Mon, 26 Oct 2026 14:07:08 -0300 Subject: [PATCH 3/6] feat(rpc): serve register_validator and forward to the builder A validator client's registrations reach the builder only for validators the head state holds as active or pending, as the beacon-APIs require; the rest are counted, not refused. Each kept registration also sets its validator's fee recipient, so a block that falls back to local building still pays the registered address. A builder's 400 is passed through and any other failure is a 500, so a misconfigured validator client hears its registrations went nowhere. --- .../src/beacon/builder_client/mock_relay.rs | 8 + crates/net/rpc/src/beacon/mod.rs | 2 + crates/net/rpc/src/beacon/registrations.rs | 330 ++++++++++++++++++ crates/net/rpc/src/metrics.rs | 20 +- docs/metrics.md | 1 + docs/rpc.md | 10 + 6 files changed, 370 insertions(+), 1 deletion(-) create mode 100644 crates/net/rpc/src/beacon/registrations.rs 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 1e0285c0..92c6d228 100644 --- a/crates/net/rpc/src/beacon/builder_client/mock_relay.rs +++ b/crates/net/rpc/src/beacon/builder_client/mock_relay.rs @@ -54,6 +54,8 @@ struct RelayState { ssz_submissions: usize, /// Whether `status` answers 503. unavailable: bool, + /// `registerValidator` calls received, whatever their outcome. + registration_calls: usize, } /// A fixed answer to every `getHeader`. @@ -165,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() } @@ -253,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) { diff --git a/crates/net/rpc/src/beacon/mod.rs b/crates/net/rpc/src/beacon/mod.rs index 0c707791..784cf163 100644 --- a/crates/net/rpc/src/beacon/mod.rs +++ b/crates/net/rpc/src/beacon/mod.rs @@ -32,6 +32,7 @@ pub(crate) mod node; pub(crate) mod operations; pub(crate) mod pool; pub(crate) mod proposal; +pub(crate) mod registrations; pub(crate) mod proposer_preferences; pub(crate) mod ptc; pub(crate) mod states; @@ -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..5d2c5b2c --- /dev/null +++ b/crates/net/rpc/src/beacon/registrations.rs @@ -0,0 +1,330 @@ +//! `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: 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::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, 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 { + Router::new().route( + "/eth/v1/validator/register_validator", + post(post_register_validator), + ) +} + +async fn post_register_validator( + State(store): State, + Extension(builder): Extension>, + Extension(fee_recipients): Extension, + headers: HeaderMap, + body: Bytes, +) -> Response { + register(&store, builder.as_ref(), &fee_recipients, &headers, &body) + .await + .map_or_else(IntoResponse::into_response, |()| { + axum::http::StatusCode::OK.into_response() + }) +} + +async fn register( + store: &Store, + builder: Option<&BuilderClient>, + fee_recipients: &FeeRecipients, + headers: &HeaderMap, + body: &[u8], +) -> Result<(), ApiError> { + let builder = + builder.ok_or(ApiError::Internal("no builder is configured (--builder)"))?; + let registrations = decode(headers, body)?; + let (_, state) = head(store)?; + + let submitted = registrations.len(); + let kept = active_or_pending(&state, registrations); + 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(); + match builder.register_validators(&kept).await { + Ok(()) => { + inc_builder_registrations("forwarded", kept.len() as u64); + Ok(()) + } + Err(err) => { + inc_builder_registrations("refused", kept.len() as u64); + warn!(%err, count = kept.len(), "Builder did not take the registrations"); + 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 +/// its own epoch, each with its validator index. One pass over the registry, +/// whatever the number of registrations. +fn active_or_pending( + state: &BeaconState, + registrations: Vec, +) -> Vec<(ValidatorIndex, SignedValidatorRegistrationV1)> { + let epoch = compute_epoch_at_slot(state.slot()); + 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)| wanted.contains(&validator.pubkey) && epoch < validator.exit_epoch) + .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::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); + } +} diff --git a/crates/net/rpc/src/metrics.rs b/crates/net/rpc/src/metrics.rs index bb1529e6..f39b8b9a 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, refused by 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/metrics.md b/docs/metrics.md index 01ac03ab..a2b6af3d 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -468,6 +468,7 @@ ethlambda-specific, not part of the leanMetrics spec. |------|------|-------|-------------------------|--------|---------| | `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`, `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\|refused\|dropped | | ### Beacon Committee Cache diff --git a/docs/rpc.md b/docs/rpc.md index 04d72589..a913deeb 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` | none | 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,15 @@ 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: 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. From 13db7ca3de340aede4f24314916f26e546c7aa74 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Mon, 26 Oct 2026 14:08:39 -0300 Subject: [PATCH 4/6] style: format the builder registration changes --- crates/net/rpc/src/beacon/mod.rs | 2 +- crates/net/rpc/src/beacon/registrations.rs | 45 ++++++++++++++++------ 2 files changed, 34 insertions(+), 13 deletions(-) diff --git a/crates/net/rpc/src/beacon/mod.rs b/crates/net/rpc/src/beacon/mod.rs index 784cf163..dc6af10c 100644 --- a/crates/net/rpc/src/beacon/mod.rs +++ b/crates/net/rpc/src/beacon/mod.rs @@ -32,9 +32,9 @@ pub(crate) mod node; pub(crate) mod operations; pub(crate) mod pool; pub(crate) mod proposal; -pub(crate) mod registrations; 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; diff --git a/crates/net/rpc/src/beacon/registrations.rs b/crates/net/rpc/src/beacon/registrations.rs index 5d2c5b2c..b24126aa 100644 --- a/crates/net/rpc/src/beacon/registrations.rs +++ b/crates/net/rpc/src/beacon/registrations.rs @@ -60,8 +60,7 @@ async fn register( headers: &HeaderMap, body: &[u8], ) -> Result<(), ApiError> { - let builder = - builder.ok_or(ApiError::Internal("no builder is configured (--builder)"))?; + let builder = builder.ok_or(ApiError::Internal("no builder is configured (--builder)"))?; let registrations = decode(headers, body)?; let (_, state) = head(store)?; @@ -78,8 +77,10 @@ async fn register( recipients.insert(*index, registration.message.fee_recipient); } } - let kept: Vec = - kept.into_iter().map(|(_, registration)| registration).collect(); + let kept: Vec = kept + .into_iter() + .map(|(_, registration)| registration) + .collect(); match builder.register_validators(&kept).await { Ok(()) => { inc_builder_registrations("forwarded", kept.len() as u64); @@ -97,7 +98,10 @@ async fn register( } /// The request body: a JSON array, or the SSZ list builder-specs defines. -fn decode(headers: &HeaderMap, body: &[u8]) -> Result, ApiError> { +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()) @@ -160,7 +164,10 @@ mod tests { 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 + .validator_mut(index as ValidatorIndex) + .unwrap() + .exit_epoch = 0; } state.apply_pending_mutations(); beacon_store_at(state).0 @@ -194,7 +201,10 @@ mod tests { } fn client_for(relay: &MockRelay) -> Option { - Some(BuilderClient::new(relay.url.clone(), Duration::from_secs(1), "ethlambda/test").unwrap()) + Some( + BuilderClient::new(relay.url.clone(), Duration::from_secs(1), "ethlambda/test") + .unwrap(), + ) } #[tokio::test] @@ -218,8 +228,14 @@ mod tests { 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))); + assert_eq!( + recipients.get(&0), + Some(&ExecutionAddress::repeat_byte(0x11)) + ); + assert_eq!( + recipients.get(&1), + Some(&ExecutionAddress::repeat_byte(0x11)) + ); } #[tokio::test] @@ -299,8 +315,12 @@ mod tests { 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 builder = BuilderClient::new( + "http://127.0.0.1:1", + Duration::from_secs(1), + "ethlambda/test", + ) + .unwrap(); let (status, _) = post( store, Some(builder), @@ -322,7 +342,8 @@ mod tests { .body(Body::from(list.to_ssz())) .unwrap(); - let (status, body) = post(store, client_for(&relay), FeeRecipients::default(), request).await; + let (status, body) = + post(store, client_for(&relay), FeeRecipients::default(), request).await; assert_eq!(status, StatusCode::OK, "{body}"); assert_eq!(relay.registrations(), registrations); From 6497d414d310539c6cd19d3076f482737f670d2c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Mon, 26 Oct 2026 14:25:01 -0300 Subject: [PATCH 5/6] fix(rpc): keep register_validator's registry walk off the runtime The head-state read and the walk over every validator now run on a blocking task, as the duty endpoints' walks do, and the forward to the builder runs on its own task so a validator client hanging up cannot cancel it. Exit is judged against the wall clock too, a malformed body is a 400 with or without a builder, the route takes the bodies a validator client sending all its keys at once needs, and a zero header timeout is refused. --- bin/ethlambda/src/cli.rs | 20 ++- crates/net/rpc/src/beacon/registrations.rs | 160 ++++++++++++++++++--- crates/net/rpc/src/metrics.rs | 2 +- docs/metrics.md | 2 +- docs/rpc.md | 2 +- 5 files changed, 165 insertions(+), 21 deletions(-) diff --git a/bin/ethlambda/src/cli.rs b/bin/ethlambda/src/cli.rs index d82daf0c..4101c56a 100644 --- a/bin/ethlambda/src/cli.rs +++ b/bin/ethlambda/src/cli.rs @@ -318,7 +318,12 @@ pub(crate) struct MainnetOptions { /// 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")] + #[arg( + long, + default_value_t = 1000, + requires = "builder", + value_parser = clap::value_parser!(u64).range(1..), + )] pub(crate) builder_header_timeout: u64, } @@ -1062,6 +1067,19 @@ mod tests { } } + #[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(); diff --git a/crates/net/rpc/src/beacon/registrations.rs b/crates/net/rpc/src/beacon/registrations.rs index b24126aa..c22f3e7c 100644 --- a/crates/net/rpc/src/beacon/registrations.rs +++ b/crates/net/rpc/src/beacon/registrations.rs @@ -12,7 +12,7 @@ use std::collections::{HashMap, HashSet}; use axum::body::Bytes; -use axum::extract::State; +use axum::extract::{DefaultBodyLimit, State}; use axum::http::HeaderMap; use axum::response::{IntoResponse, Response}; use axum::routing::post; @@ -22,7 +22,7 @@ use ethlambda_types::beacon::builder::{ SignedValidatorRegistrationV1, SignedValidatorRegistrations, }; use ethlambda_types::beacon::containers::BeaconState; -use ethlambda_types::beacon::primitives::{BlsPubkey, ValidatorIndex}; +use ethlambda_types::beacon::primitives::{BlsPubkey, Epoch, ValidatorIndex}; use ethlambda_types::beacon::signing::compute_epoch_at_slot; use libssz::SszDecode as _; use tracing::warn; @@ -33,9 +33,11 @@ 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), + post(post_register_validator).layer(DefaultBodyLimit::max(super::MAX_PUBLISH_BODY_BYTES)), ) } @@ -46,7 +48,7 @@ async fn post_register_validator( headers: HeaderMap, body: Bytes, ) -> Response { - register(&store, builder.as_ref(), &fee_recipients, &headers, &body) + register(store, builder, fee_recipients, &headers, &body) .await .map_or_else(IntoResponse::into_response, |()| { axum::http::StatusCode::OK.into_response() @@ -54,18 +56,29 @@ async fn post_register_validator( } async fn register( - store: &Store, - builder: Option<&BuilderClient>, - fee_recipients: &FeeRecipients, + store: Store, + builder: Option, + fee_recipients: FeeRecipients, headers: &HeaderMap, body: &[u8], ) -> Result<(), ApiError> { - let builder = builder.ok_or(ApiError::Internal("no builder is configured (--builder)"))?; + // Decoded first, so a malformed body is a 400 whether or not a builder is + // configured. let registrations = decode(headers, body)?; - let (_, state) = head(store)?; + 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 = active_or_pending(&state, registrations); + 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(()); @@ -81,14 +94,20 @@ async fn register( .into_iter() .map(|(_, registration)| registration) .collect(); - match builder.register_validators(&kept).await { + // On its own task: a validator client that hangs up drops this handler, + // and the forward must complete regardless. + let count = kept.len() as u64; + let forwarded = tokio::spawn(async move { builder.register_validators(&kept).await }) + .await + .map_err(|_| ApiError::Internal("forwarding the registrations failed"))?; + match forwarded { Ok(()) => { - inc_builder_registrations("forwarded", kept.len() as u64); + inc_builder_registrations("forwarded", count); Ok(()) } Err(err) => { - inc_builder_registrations("refused", kept.len() as u64); - warn!(%err, count = kept.len(), "Builder did not take the registrations"); + inc_builder_registrations("failed", count); + warn!(%err, count, "Builder did not take the registrations"); Err(match err { BuilderError::BadRequest(message) => ApiError::BadRequestDetail(message), _ => ApiError::Internal("the builder did not take the registrations"), @@ -112,13 +131,14 @@ fn decode( } /// The registrations whose validator `state` holds as active or pending at -/// its own epoch, each with its validator index. One pass over the registry, +/// 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()); + let epoch = compute_epoch_at_slot(state.slot()).max(wall_epoch); let wanted: HashSet = registrations .iter() .map(|registration| registration.message.pubkey) @@ -128,7 +148,7 @@ fn active_or_pending( let indices: HashMap = state .iter_validators() .enumerate() - .filter(|(_, validator)| wanted.contains(&validator.pubkey) && epoch < validator.exit_epoch) + .filter(|(_, validator)| epoch < validator.exit_epoch && wanted.contains(&validator.pubkey)) .map(|(index, validator)| (validator.pubkey, index as ValidatorIndex)) .collect(); registrations @@ -147,6 +167,7 @@ mod tests { 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 _; @@ -348,4 +369,109 @@ mod tests { 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}"); + } } diff --git a/crates/net/rpc/src/metrics.rs b/crates/net/rpc/src/metrics.rs index f39b8b9a..14baad83 100644 --- a/crates/net/rpc/src/metrics.rs +++ b/crates/net/rpc/src/metrics.rs @@ -69,7 +69,7 @@ pub(crate) async fn get_metrics() -> impl IntoResponse { } /// Count `register_validator` entries by what became of them: forwarded to -/// the builder, refused by it, or dropped here as an unknown or exited +/// 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(|| { diff --git a/docs/metrics.md b/docs/metrics.md index a2b6af3d..5e3edce4 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -468,7 +468,7 @@ ethlambda-specific, not part of the leanMetrics spec. |------|------|-------|-------------------------|--------|---------| | `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`, `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\|refused\|dropped | | +| `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 a913deeb..1f92fbc4 100644 --- a/docs/rpc.md +++ b/docs/rpc.md @@ -274,7 +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` | none | Forwards builder registrations of active or pending validators to `--builder`, and sets their fee recipients | +| `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 From afba2c87de8f37053d306a6bf5cbb9e8ff45c78e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Mon, 26 Oct 2026 14:27:40 -0300 Subject: [PATCH 6/6] fix(rpc): count a registration forward even if the client hangs up The outcome counter and its log move into the task that outlives the request, and a test pins exit to the wall clock as well as the head. --- crates/net/rpc/src/beacon/registrations.rs | 68 +++++++++++++++------- docs/rpc.md | 4 +- 2 files changed, 50 insertions(+), 22 deletions(-) diff --git a/crates/net/rpc/src/beacon/registrations.rs b/crates/net/rpc/src/beacon/registrations.rs index c22f3e7c..bca74807 100644 --- a/crates/net/rpc/src/beacon/registrations.rs +++ b/crates/net/rpc/src/beacon/registrations.rs @@ -3,7 +3,9 @@ //! builder `--builder` names. //! //! Validators that the head state does not hold, or holds as exited, are -//! dropped and counted rather than refused: the beacon-APIs say they must +//! 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 @@ -95,25 +97,25 @@ async fn register( .map(|(_, registration)| registration) .collect(); // On its own task: a validator client that hangs up drops this handler, - // and the forward must complete regardless. - let count = kept.len() as u64; - let forwarded = tokio::spawn(async move { builder.register_validators(&kept).await }) - .await - .map_err(|_| ApiError::Internal("forwarding the registrations failed"))?; - match forwarded { - Ok(()) => { - inc_builder_registrations("forwarded", count); - Ok(()) - } - Err(err) => { - inc_builder_registrations("failed", count); - warn!(%err, count, "Builder did not take the registrations"); - Err(match err { - BuilderError::BadRequest(message) => ApiError::BadRequestDetail(message), - _ => ApiError::Internal("the builder did not take the registrations"), - }) + // 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. @@ -131,8 +133,8 @@ fn decode( } /// 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. +/// 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, @@ -474,4 +476,28 @@ mod tests { 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/docs/rpc.md b/docs/rpc.md index 1f92fbc4..85e8aaea 100644 --- a/docs/rpc.md +++ b/docs/rpc.md @@ -413,7 +413,9 @@ the chain actor writes, so no request waits on the actor. - **`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: the beacon-APIs say + (`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