From 85b1f2ef44f03d94175af6c3ebafb75333908032 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/8] 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 01e50834a..1e0285c0f 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 4ebf7ca87..d702e3d03 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 596913767..b2874c536 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 48616f7b4..01ac03aba 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 d7e0b1b3df6b04236f73784cbca3704531aa74c0 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/8] 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 4dbe91741..d82daf0c8 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 9110da5de..813055209 100644 --- a/bin/ethlambda/src/main.rs +++ b/bin/ethlambda/src/main.rs @@ -448,6 +448,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 @@ -842,6 +871,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 2fb202871..af3f5f9cc 100644 --- a/crates/net/rpc/src/lib.rs +++ b/crates/net/rpc/src/lib.rs @@ -242,6 +242,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. @@ -267,6 +270,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 2dcbf6703..489b24f9e 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 b52d7803ce5def6ff45c1ee974e240144210b1c6 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/8] 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 1e0285c0f..92c6d2281 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 0c7077911..784cf163a 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 000000000..5d2c5b2ce --- /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 bb1529e63..f39b8b9a4 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 01ac03aba..a2b6af3da 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 bdcc260ec..f5824f67a 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 ce9b74cd54c10a06277b10789213abf6844ae51b 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/8] 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 784cf163a..dc6af10c5 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 5d2c5b2ce..b24126aa9 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 8e240979058e716faad8c7406963e76d73c72f40 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/8] 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 d82daf0c8..4101c56a5 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 b24126aa9..c22f3e7c9 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 f39b8b9a4..14baad83e 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 a2b6af3da..5e3edce48 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 f5824f67a..9193b305b 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 19ba324c604975d3bd8fbaaf24b7a62e81e9563e 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/8] 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 c22f3e7c9..bca74807e 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 9193b305b..e63a27332 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 From 6fb5e57a7d3b7779dd019bfbfb6b08be1c6766f4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Thu, 5 Nov 2026 15:17:34 -0300 Subject: [PATCH 7/8] fix(rpc): address the register_validator review - Record fee recipients before looking for a builder, so they hold with no --builder too, as Lighthouse does. - Keep one registration per validator: its newest by timestamp, the later one on a tie, as builders order them. The rest are counted as `duplicate` in lean_beacon_builder_registrations_total. - Answer an empty list without reading the head state (still a 500 with no builder). - Stop following builder redirects unless --builder-follow-redirects is set. A refused redirect fails the call with an error naming its target, minus any user or password, and is counted as `redirected`. - Check --builder's URL while the flags are parsed. - Reword the startup status log: nothing waits on that probe. - Tests: duplicate registrations and timestamp ties, empty lists with and without a builder, a slashed validator that has not exited, the fee recipient after prepare_beacon_proposer and register_validator in turn, and redirects refused, followed and shown without credentials. Unreachable builders use a released local port instead of port 1, and the pending-validator test pins the wall clock to slot 0. - Note the per-call registry walk as a future improvement. Claude-Session: https://claude.ai/code/session_011vghNjQ8uikd9KL5Jo5at7 --- bin/ethlambda/src/cli.rs | 55 +++- bin/ethlambda/src/main.rs | 7 +- .../rpc/src/beacon/builder_client/error.rs | 36 +++ .../src/beacon/builder_client/mock_relay.rs | 48 ++- .../net/rpc/src/beacon/builder_client/mod.rs | 56 +++- .../rpc/src/beacon/builder_client/tests.rs | 61 +++- crates/net/rpc/src/beacon/registrations.rs | 277 ++++++++++++++++-- crates/net/rpc/src/metrics.rs | 5 +- docs/cli.md | 1 + docs/metrics.md | 4 +- docs/rpc.md | 16 +- 11 files changed, 518 insertions(+), 48 deletions(-) diff --git a/bin/ethlambda/src/cli.rs b/bin/ethlambda/src/cli.rs index 4101c56a5..27bb92bee 100644 --- a/bin/ethlambda/src/cli.rs +++ b/bin/ethlambda/src/cli.rs @@ -311,10 +311,17 @@ pub(crate) struct MainnetOptions { /// 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)] + /// client configured for a builder hears that it has none (the fee + /// recipients it names are still recorded). + #[arg(long, value_parser = parse_builder_url)] pub(crate) builder: Option, + /// Follow HTTP redirects from the builder. Off by default, so registrations + /// and signed blocks go only to the URL `--builder` names: a redirect fails + /// the call, with an error naming where it pointed. + #[arg(long, requires = "builder")] + pub(crate) builder_follow_redirects: bool, + /// 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. @@ -327,6 +334,15 @@ pub(crate) struct MainnetOptions { pub(crate) builder_header_timeout: u64, } +/// `--builder`'s value, refused while the flags are parsed unless it is an +/// http or https URL: the same check the builder client makes, so the error +/// names the flag instead of surfacing at startup. +fn parse_builder_url(value: &str) -> Result { + ethlambda_rpc::BuilderClient::check_url(value) + .map(|()| value.to_owned()) + .map_err(|err| err.to_string()) +} + /// Flags only the lean chain takes. #[derive(Debug, clap::Args)] pub(crate) struct LeanOptions { @@ -1088,6 +1104,41 @@ mod tests { try_parse_from(args).expect_err("a timeout without a builder must not parse"); } + #[test] + fn a_builder_url_is_checked_while_parsing() { + for bad in ["not a url", "localhost:18550"] { + let mut args = beacon_args(); + args.extend(["--builder", bad]); + + let err = try_parse_from(args).expect_err("a bad builder url must not parse"); + + assert!(err.to_string().contains("--builder"), "{err}"); + } + } + + #[test] + fn redirects_are_off_unless_asked_for() { + let mut args = beacon_args(); + args.extend(["--builder", "http://127.0.0.1:18550"]); + assert!(!parse_beacon(args).mainnet.builder_follow_redirects); + + let mut args = beacon_args(); + args.extend([ + "--builder", + "http://127.0.0.1:18550", + "--builder-follow-redirects", + ]); + assert!(parse_beacon(args).mainnet.builder_follow_redirects); + } + + #[test] + fn following_redirects_needs_a_builder() { + let mut args = beacon_args(); + args.push("--builder-follow-redirects"); + + try_parse_from(args).expect_err("following redirects 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 813055209..74747f7a3 100644 --- a/bin/ethlambda/src/main.rs +++ b/bin/ethlambda/src/main.rs @@ -457,14 +457,17 @@ async fn run_node(options: Options) -> eyre::Result<()> { .builder .as_deref() .map(|url| { - ethlambda_rpc::BuilderClient::new( + ethlambda_rpc::BuilderClient::with_redirects( url, std::time::Duration::from_millis(mainnet.builder_header_timeout), version::client_version(), + mainnet.builder_follow_redirects, ) }) .transpose()?, }; + // A one-off probe, for the log only: nothing waits on it or reads its + // answer, and every later call to the builder is tried on its own. if let Some(builder) = builder.clone() { tokio::spawn(async move { let _ = builder @@ -472,7 +475,7 @@ async fn run_node(options: Options) -> eyre::Result<()> { .await .inspect(|()| info!("Builder is up")) .inspect_err(|err| { - warn!(%err, "Builder is not answering; proposals build locally until it does") + warn!(%err, "Builder did not answer the startup status check; each call to it is still tried") }); }); } diff --git a/crates/net/rpc/src/beacon/builder_client/error.rs b/crates/net/rpc/src/beacon/builder_client/error.rs index 6da14b3c1..3d4463c93 100644 --- a/crates/net/rpc/src/beacon/builder_client/error.rs +++ b/crates/net/rpc/src/beacon/builder_client/error.rs @@ -3,6 +3,7 @@ use std::time::Duration; use reqwest::StatusCode; +use reqwest::header; use super::{read_body, truncate_error_text}; @@ -26,6 +27,12 @@ pub enum BuilderError { /// A status the call does not expect. #[error("builder answered {status}: {body}")] Status { status: u16, body: String }, + /// A redirect, which the client does not follow unless told to. + #[error( + "builder redirected the request to {location} ({status}); redirects are not \ + followed unless --builder-follow-redirects is set" + )] + Redirected { status: u16, location: String }, /// A body that does not decode as what the call returns. #[error("builder response did not decode: {0}")] Decode(String), @@ -60,8 +67,15 @@ impl BuilderError { /// The error a status other than the call's success stands for. A 400 /// carries the builder's `message`, or its raw body if it sent no JSON. + /// A redirect carries its target. pub(super) async fn from_status(response: reqwest::Response, deadline: Duration) -> Self { let status = response.status(); + if status.is_redirection() { + return Self::Redirected { + status: status.as_u16(), + location: shown_location(&response), + }; + } let body = read_body(response, deadline).await.unwrap_or_default(); let body = String::from_utf8_lossy(&body).into_owned(); if status != StatusCode::BAD_REQUEST { @@ -83,6 +97,7 @@ impl BuilderError { Self::Transport(_) => "unreachable", Self::BadRequest(_) => "bad_request", Self::Status { .. } => "unexpected_status", + Self::Redirected { .. } => "redirected", Self::Decode(_) | Self::Incomplete(_) | Self::TooLarge { .. } => "invalid_response", Self::InvalidUrl(_) => "invalid_url", Self::UnsupportedFork(_) => "unsupported_fork", @@ -91,6 +106,27 @@ impl BuilderError { } } +/// A redirect's target as an error shows it: without any user or password, +/// since a URL may carry credentials (see `from_reqwest`), and cut to the +/// error text cap. +fn shown_location(response: &reqwest::Response) -> String { + let Some(value) = response.headers().get(header::LOCATION) else { + return "nowhere (no Location header)".to_owned(); + }; + let location = String::from_utf8_lossy(value.as_bytes()).into_owned(); + let location = match reqwest::Url::parse(&location) { + Ok(mut url) if !url.username().is_empty() || url.password().is_some() => { + // Neither fails on a URL that already has a user or password. + let _ = url.set_username(""); + let _ = url.set_password(None); + url.to_string() + } + // A relative target, or one with nothing to hide, is shown as sent. + _ => location, + }; + truncate_error_text(location) +} + /// builder-specs' `ErrorMessage`, of which only the message is kept. #[derive(serde::Deserialize)] struct ErrorMessage { 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 92c6d2281..f8f1e16de 100644 --- a/crates/net/rpc/src/beacon/builder_client/mock_relay.rs +++ b/crates/net/rpc/src/beacon/builder_client/mock_relay.rs @@ -6,15 +6,17 @@ //! 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`. +//! It serves `status` with 200, or 503 after `make_unavailable`, and answers +//! every request with a 307 after `redirect_to`. use std::sync::{Arc, Mutex}; use std::time::Duration; use axum::Router; use axum::body::Bytes; -use axum::extract::{Path, State}; +use axum::extract::{Path, Request, State}; use axum::http::{HeaderMap, HeaderValue, StatusCode, header}; +use axum::middleware::{self, Next}; use axum::response::{IntoResponse, Response}; use axum::routing::{get, post}; use ethlambda_state_transition::beacon::bls; @@ -56,6 +58,9 @@ struct RelayState { unavailable: bool, /// `registerValidator` calls received, whatever their outcome. registration_calls: usize, + /// Where every request is sent instead, with a 307: this base URL + /// followed by the request's own path. + redirect_to: Option, } /// A fixed answer to every `getHeader`. @@ -98,6 +103,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)) + .layer(middleware::from_fn_with_state(state.clone(), redirect)) .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()); @@ -159,6 +165,13 @@ impl MockRelay { self.state.lock().unwrap().unavailable = true; } + /// Answer every request with a 307 to `base` followed by the request's + /// own path, as a builder that moved does. A 307 keeps the method and the + /// body, so a client that follows it lands a POST on `base` intact. + pub(crate) fn redirect_to(&self, base: &str) { + self.state.lock().unwrap().redirect_to = Some(base.to_owned()); + } + pub(crate) fn blinded_blocks(&self) -> Vec { self.state.lock().unwrap().blinded_blocks.clone() } @@ -199,6 +212,19 @@ impl MockRelay { } } +async fn redirect(State(state): State, request: Request, next: Next) -> Response { + let target = state.lock().unwrap().redirect_to.clone(); + let Some(base) = target else { + return next.run(request).await; + }; + let location = format!("{base}{}", request.uri().path()); + ( + StatusCode::TEMPORARY_REDIRECT, + [(header::LOCATION, location)], + ) + .into_response() +} + async fn header( State(state): State, Path((slot, parent_hash, pubkey)): Path<(String, String, String)>, @@ -330,6 +356,16 @@ pub(crate) fn signed_registration( timestamp: 1_700_000_000, pubkey: BlsPubkey(secret_key_for(index).sk_to_pk().to_bytes()), }; + sign_registration(index, message, genesis_fork_version) +} + +/// `message` signed with test validator `index`'s key, for registrations +/// that differ from [`signed_registration`]'s in timestamp or fee recipient. +pub(crate) fn sign_registration( + index: usize, + message: ValidatorRegistrationV1, + genesis_fork_version: Version, +) -> SignedValidatorRegistrationV1 { let domain = compute_builder_domain(genesis_fork_version); let root = compute_signing_root(message.hash_tree_root(), domain); SignedValidatorRegistrationV1 { @@ -338,6 +374,14 @@ pub(crate) fn signed_registration( } } +/// A local URL nothing listens on: its port was bound and released, so a +/// connection is refused at once. Port 1 usually is too, but a filtered +/// port 1 would make a test wait for the call's timeout instead. +pub(crate) fn unreachable_url() -> String { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + format!("http://{}", listener.local_addr().unwrap()) +} + fn is_ssz_request(headers: &HeaderMap) -> bool { headers .get(header::CONTENT_TYPE) diff --git a/crates/net/rpc/src/beacon/builder_client/mod.rs b/crates/net/rpc/src/beacon/builder_client/mod.rs index d702e3d03..d3d00f637 100644 --- a/crates/net/rpc/src/beacon/builder_client/mod.rs +++ b/crates/net/rpc/src/beacon/builder_client/mod.rs @@ -14,6 +14,10 @@ //! 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. //! +//! Redirects are not followed unless `--builder-follow-redirects` asks for +//! it, so the node sends registrations and signed blocks only to the URL the +//! operator named. A refused redirect is an error naming its target. +//! //! `status` is a liveness probe the node calls once at startup. //! //! Every call is timed into `lean_beacon_builder_request_seconds`, by call @@ -31,6 +35,7 @@ use ethlambda_types::beacon::primitives::{BlsPubkey, ExecutionBlockHash, Slot}; use libssz::{SszDecode as _, SszEncode as _}; use reqwest::StatusCode; use reqwest::header::{self, HeaderMap}; +use reqwest::redirect::Policy; use tracing::info; use crate::metrics::observe_builder_request; @@ -87,7 +92,8 @@ pub struct VersionedBid { } impl BuilderClient { - /// A client for `base_url` that identifies itself as `user_agent`. + /// A client for `base_url` that identifies itself as `user_agent` and + /// does not follow redirects (see [`with_redirects`](Self::with_redirects)). /// /// `header_timeout` bounds `getHeader`. It is also sent to the builder as /// `X-Timeout-Ms`, so the builder can hold its answer for as long as the @@ -96,19 +102,33 @@ impl BuilderClient { base_url: impl Into, header_timeout: Duration, user_agent: &str, + ) -> Result { + Self::with_redirects(base_url, header_timeout, user_agent, false) + } + + /// [`new`](Self::new), but following up to ten redirects (reqwest's + /// default) when `follow_redirects` is set. + /// + /// Without it every call goes exactly where `base_url` points. A redirect + /// fails the call as [`BuilderError::Redirected`], which names the target, + /// rather than carrying registrations or a signed block somewhere + /// `--builder` does not name. + pub fn with_redirects( + base_url: impl Into, + header_timeout: Duration, + user_agent: &str, + follow_redirects: bool, ) -> Result { let base_url = base_url.into(); - let url = reqwest::Url::parse(&base_url) - .map_err(|err| BuilderError::InvalidUrl(err.to_string()))?; - // `localhost:18550` parses, with `localhost` as its scheme. - if !matches!(url.scheme(), "http" | "https") { - return Err(BuilderError::InvalidUrl(format!( - "expected an http or https url, got scheme {:?}", - url.scheme() - ))); - } + Self::check_url(&base_url)?; + let redirects = if follow_redirects { + Policy::default() + } else { + Policy::none() + }; let http = reqwest::Client::builder() .user_agent(user_agent) + .redirect(redirects) .build() .map_err(|err| BuilderError::Transport(err.without_url().to_string()))?; let base_url = base_url.trim_end_matches('/').to_owned(); @@ -120,6 +140,22 @@ impl BuilderClient { }) } + /// Whether `base_url` is a URL a client can be built for: one that + /// parses, with an http or https scheme. The CLI checks `--builder` with + /// this, so a bad value is refused while the flags are parsed. + pub fn check_url(base_url: &str) -> Result<(), BuilderError> { + let url = reqwest::Url::parse(base_url) + .map_err(|err| BuilderError::InvalidUrl(err.to_string()))?; + // `localhost:18550` parses, with `localhost` as its scheme. + if !matches!(url.scheme(), "http" | "https") { + return Err(BuilderError::InvalidUrl(format!( + "expected an http or https url, got scheme {:?}", + url.scheme() + ))); + } + Ok(()) + } + /// `getHeader`: the builder's best bid for `slot`, on top of /// `parent_hash`, for the proposer `pubkey`. `None` when it has none /// (204). diff --git a/crates/net/rpc/src/beacon/builder_client/tests.rs b/crates/net/rpc/src/beacon/builder_client/tests.rs index b2874c536..9cbe671d2 100644 --- a/crates/net/rpc/src/beacon/builder_client/tests.rs +++ b/crates/net/rpc/src/beacon/builder_client/tests.rs @@ -10,7 +10,7 @@ use ethlambda_types::beacon::primitives::{ BlsPubkey, BlsSignature, ExecutionBlockHash, Root, Uint256, Version, }; -use super::mock_relay::{MockRelay, signed_registration}; +use super::mock_relay::{MockRelay, signed_registration, unreachable_url}; use super::{BuilderClient, BuilderError, VersionedBid}; /// Any version will do: the relay and the signers only have to agree on it. @@ -169,14 +169,69 @@ async fn an_ssz_bid_must_name_its_fork() { #[tokio::test] async fn an_unreachable_relay_is_reported_as_such() { - let client = - BuilderClient::new("http://127.0.0.1:1", HEADER_TIMEOUT, "ethlambda/test").unwrap(); + let client = BuilderClient::new(unreachable_url(), HEADER_TIMEOUT, "ethlambda/test").unwrap(); assert!(matches!( get_header(&client).await, Err(BuilderError::Transport(_)) )); } +#[tokio::test] +async fn a_redirect_is_refused_and_names_its_target() { + let moved = MockRelay::start(GENESIS_FORK_VERSION).await; + let target = MockRelay::start(GENESIS_FORK_VERSION).await; + moved.redirect_to(&target.url); + let registrations = vec![signed_registration(0, GENESIS_FORK_VERSION)]; + + let err = client_for(&moved) + .register_validators(®istrations) + .await + .unwrap_err(); + + let location = format!("{}/eth/v1/builder/validators", target.url); + assert!( + matches!(&err, BuilderError::Redirected { status: 307, location: shown } if *shown == location), + "{err:?}" + ); + let message = err.to_string(); + assert!(message.contains(&location), "{message}"); + assert!(message.contains("--builder-follow-redirects"), "{message}"); + assert_eq!(err.label(), "redirected"); + assert_eq!(target.registration_calls(), 0); +} + +#[tokio::test] +async fn a_redirect_is_followed_when_asked() { + let moved = MockRelay::start(GENESIS_FORK_VERSION).await; + let target = MockRelay::start(GENESIS_FORK_VERSION).await; + moved.redirect_to(&target.url); + let client = + BuilderClient::with_redirects(moved.url.clone(), HEADER_TIMEOUT, "ethlambda/test", true) + .unwrap(); + let registrations = vec![signed_registration(0, GENESIS_FORK_VERSION)]; + + client.register_validators(®istrations).await.unwrap(); + + // A 307 keeps the POST and its body, so the target got them whole. + assert_eq!(target.registrations(), registrations); +} + +#[tokio::test] +async fn a_redirect_target_is_shown_without_its_credentials() { + let moved = MockRelay::start(GENESIS_FORK_VERSION).await; + moved.redirect_to("http://operator:hunter2@127.0.0.1:9"); + + let err = client_for(&moved).status().await.unwrap_err(); + + let message = err.to_string(); + assert!( + message.contains("http://127.0.0.1:9/eth/v1/builder/status"), + "{message}" + ); + assert!(!message.contains("hunter2"), "{message}"); + assert!(!message.contains("operator"), "{message}"); +} + const JSON: &str = "application/json"; const TEXT: &str = "text/plain"; const SSZ: &str = "application/octet-stream"; diff --git a/crates/net/rpc/src/beacon/registrations.rs b/crates/net/rpc/src/beacon/registrations.rs index bca74807e..3ec1e07bc 100644 --- a/crates/net/rpc/src/beacon/registrations.rs +++ b/crates/net/rpc/src/beacon/registrations.rs @@ -9,7 +9,14 @@ //! 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. +//! all still pays the address the validator registered. As in Lighthouse, +//! that holds with no builder configured (the answer is still a 500) and +//! whatever the builder answers. Signatures are not verified here, also as +//! in Lighthouse: the builder verifies them. +//! +//! A validator registered more than once in one request keeps only its +//! newest registration, and an empty request is answered without reading +//! the head state. use std::collections::{HashMap, HashSet}; @@ -57,6 +64,9 @@ async fn post_register_validator( }) } +/// What a validator client hears when this node has no builder to forward to. +const NO_BUILDER: &str = "no builder is configured (--builder)"; + async fn register( store: Store, builder: Option, @@ -67,11 +77,21 @@ async fn register( // 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)"))?; + // Nothing to record or forward, so no registry walk: a stream of empty + // bodies would otherwise tie up the blocking pool the duty endpoints + // share. Still a 500 without a builder, as any other body is. + if registrations.is_empty() { + builder.ok_or(ApiError::Internal(NO_BUILDER))?; + return Ok(()); + } + + let submitted = registrations.len(); + let registrations = newest_per_validator(registrations); + inc_builder_registrations("duplicate", (submitted - registrations.len()) as u64); // 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 candidates = 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 @@ -81,17 +101,22 @@ async fn register( }) .await .map_err(|_| ApiError::Internal("checking the registrations failed"))??; - inc_builder_registrations("dropped", (submitted - kept.len()) as u64); - if kept.is_empty() { - return Ok(()); - } + inc_builder_registrations("dropped", (candidates - kept.len()) as u64); + // Recorded before the builder is even looked for, as Lighthouse records + // them: they hold with no builder configured and whatever the builder + // answers. { let mut recipients = fee_recipients.lock().expect("fee recipient lock poisoned"); for (index, registration) in &kept { recipients.insert(*index, registration.message.fee_recipient); } } + let builder = builder.ok_or(ApiError::Internal(NO_BUILDER))?; + if kept.is_empty() { + return Ok(()); + } + let kept: Vec = kept .into_iter() .map(|(_, registration)| registration) @@ -132,9 +157,46 @@ fn decode( } } +/// One registration per validator: its newest by `timestamp`, the later one +/// in the request on a tie, which is the one a builder keeps (builder-specs' +/// `process_registration` takes any registration not older than the one it +/// holds). The kept ones stay in request order. +/// +/// So one validator repeated many times in a large body is forwarded once, +/// and its fee recipient is the newest registration's, whatever the order. +fn newest_per_validator( + registrations: Vec, +) -> Vec { + let mut newest: HashMap = HashMap::with_capacity(registrations.len()); + for (position, registration) in registrations.iter().enumerate() { + let message = ®istration.message; + newest + .entry(message.pubkey) + .and_modify(|held| { + if message.timestamp >= registrations[*held].message.timestamp { + *held = position; + } + }) + .or_insert(position); + } + let mut keep = vec![false; registrations.len()]; + for position in newest.into_values() { + keep[position] = true; + } + registrations + .into_iter() + .zip(keep) + .filter_map(|(registration, keep)| keep.then_some(registration)) + .collect() +} + /// 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. +/// +/// Future improvement: that pass happens on every call, over the whole +/// registry, with nothing bounding how many run at once. A pubkey-to-index +/// map kept beside the head state would make it one lookup per registration. fn active_or_pending( state: &BeaconState, registrations: Vec, @@ -177,8 +239,10 @@ mod tests { use tower::ServiceExt as _; use super::*; - use crate::beacon::builder_client::mock_relay::{MockRelay, signed_registration}; - use crate::test_utils::beacon_store_at; + use crate::beacon::builder_client::mock_relay::{ + MockRelay, sign_registration, signed_registration, unreachable_url, + }; + use crate::test_utils::{beacon_store_at, beacon_store_at_clock}; const VALIDATORS: usize = 8; @@ -223,6 +287,11 @@ mod tests { .unwrap() } + /// A builder on a local port nothing listens on. + fn unreachable_builder() -> BuilderClient { + BuilderClient::new(unreachable_url(), Duration::from_secs(1), "ethlambda/test").unwrap() + } + fn client_for(relay: &MockRelay) -> Option { Some( BuilderClient::new(relay.url.clone(), Duration::from_secs(1), "ethlambda/test") @@ -338,12 +407,7 @@ 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 = unreachable_builder(); let (status, _) = post( store, Some(builder), @@ -374,13 +438,18 @@ mod tests { /// A store whose validator `index` is pending: its activation is in the /// future and its exit is `FAR_FUTURE_EPOCH`. + /// + /// The wall clock is pinned to slot 0, as the head is. `beacon_store_at` + /// puts it at mainnet's current epoch instead, where an activation at + /// epoch 1000 is long past and the validator would be active, not + /// pending. 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 + beacon_store_at_clock(state, 0).0 } #[tokio::test] @@ -428,12 +497,7 @@ mod tests { 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 builder = unreachable_builder(); let fee_recipients = FeeRecipients::default(); let (status, _) = post( @@ -451,6 +515,27 @@ mod tests { ); } + #[tokio::test] + async fn fee_recipients_are_set_even_without_a_builder() { + let store = store_with_exited(&[]); + let registration = signed_registration(0, genesis_fork_version(&store)); + let fee_recipients = FeeRecipients::default(); + + let (status, _) = post( + store, + None, + 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(&[]); @@ -500,4 +585,152 @@ mod tests { assert_eq!(status, StatusCode::OK, "{body}"); assert_eq!(relay.registrations(), [kept]); } + + /// Validator `index`'s registration at `timestamp`, paying an address of + /// `fee_byte` repeated, signed so the relay takes it. + fn registration_at( + store: &Store, + index: usize, + timestamp: u64, + fee_byte: u8, + ) -> SignedValidatorRegistrationV1 { + let version = genesis_fork_version(store); + let mut message = signed_registration(index, version).message; + message.timestamp = timestamp; + message.fee_recipient = ExecutionAddress::repeat_byte(fee_byte); + sign_registration(index, message, version) + } + + #[tokio::test] + async fn only_the_newest_registration_of_a_validator_is_kept() { + let store = store_with_exited(&[]); + let relay = MockRelay::start(genesis_fork_version(&store)).await; + let oldest = registration_at(&store, 0, 100, 0x22); + let newest = registration_at(&store, 0, 300, 0x33); + let older = registration_at(&store, 0, 200, 0x44); + let other = 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(&[oldest, newest.clone(), older, other.clone()]), + ) + .await; + + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(relay.registrations(), [newest, other]); + assert_eq!( + fee_recipients.lock().unwrap().get(&0), + Some(&ExecutionAddress::repeat_byte(0x33)) + ); + } + + #[tokio::test] + async fn on_a_timestamp_tie_the_later_registration_wins() { + let store = store_with_exited(&[]); + let relay = MockRelay::start(genesis_fork_version(&store)).await; + let first = registration_at(&store, 0, 100, 0x22); + let second = registration_at(&store, 0, 100, 0x33); + + let (status, body) = post( + store, + client_for(&relay), + FeeRecipients::default(), + json_request(&[first, second.clone()]), + ) + .await; + + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(relay.registrations(), [second]); + } + + #[tokio::test] + async fn an_empty_list_is_a_success_without_a_call() { + let store = store_with_exited(&[]); + let relay = MockRelay::start(genesis_fork_version(&store)).await; + + let (status, body) = post( + store, + client_for(&relay), + FeeRecipients::default(), + json_request(&[]), + ) + .await; + + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(relay.registration_calls(), 0); + } + + #[tokio::test] + async fn an_empty_list_without_a_builder_is_a_server_error() { + let store = store_with_exited(&[]); + let (status, _) = post(store, None, FeeRecipients::default(), json_request(&[])).await; + assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR); + } + + #[tokio::test] + async fn a_slashed_validator_that_has_not_exited_is_kept() { + // Slashing sets an exit epoch; until the clock reaches it the + // validator is `active_slashed`, which builder-specs counts as active. + let mut state = with_signing_validators_at(ForkName::Fulu, VALIDATORS); + let validator = state.validator_mut(5).unwrap(); + validator.slashed = true; + validator.exit_epoch = 10; + validator.withdrawable_epoch = 8_192; + state.apply_pending_mutations(); + let (store, _) = beacon_store_at_clock(state, 0); + let relay = MockRelay::start(genesis_fork_version(&store)).await; + let slashed = signed_registration(5, genesis_fork_version(&store)); + + let (status, body) = post( + store, + client_for(&relay), + FeeRecipients::default(), + json_request(std::slice::from_ref(&slashed)), + ) + .await; + + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(relay.registrations(), [slashed]); + } + + #[tokio::test] + async fn the_last_fee_recipient_written_wins() { + let store = store_with_exited(&[]); + let relay = MockRelay::start(genesis_fork_version(&store)).await; + let registration = signed_registration(0, genesis_fork_version(&store)); + let fee_recipients = FeeRecipients::default(); + let app = crate::beacon::validator::routes() + .merge(routes()) + .with_state(store) + .layer(Extension(client_for(&relay))) + .layer(Extension(fee_recipients.clone())); + let prepare = || { + let body = serde_json::json!([ + { "validator_index": "0", "fee_recipient": format!("0x{}", "ab".repeat(20)) } + ]); + Request::post("/eth/v1/validator/prepare_beacon_proposer") + .header("content-type", "application/json") + .body(Body::from(body.to_string())) + .unwrap() + }; + let recipient = || fee_recipients.lock().unwrap().get(&0).copied(); + let prepared = ExecutionAddress::repeat_byte(0xab); + let registered = ExecutionAddress::repeat_byte(0x11); + + let response = app.clone().oneshot(prepare()).await.unwrap(); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(recipient(), Some(prepared)); + + let request = json_request(&[registration]); + let response = app.clone().oneshot(request).await.unwrap(); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(recipient(), Some(registered)); + + let response = app.oneshot(prepare()).await.unwrap(); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(recipient(), Some(prepared)); + } } diff --git a/crates/net/rpc/src/metrics.rs b/crates/net/rpc/src/metrics.rs index 14baad83e..d2d7c123a 100644 --- a/crates/net/rpc/src/metrics.rs +++ b/crates/net/rpc/src/metrics.rs @@ -69,8 +69,9 @@ pub(crate) async fn get_metrics() -> impl IntoResponse { } /// 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. +/// the builder, failed at it, dropped here as an unknown or exited validator, +/// or superseded by a newer registration of the same validator in the same +/// request (`duplicate`). pub(crate) fn inc_builder_registrations(outcome: &str, count: u64) { static LEAN_BEACON_BUILDER_REGISTRATIONS_TOTAL: LazyLock = LazyLock::new(|| { register_int_counter_vec!( diff --git a/docs/cli.md b/docs/cli.md index 489b24f9e..38351f903 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -116,6 +116,7 @@ from a normal build. | `--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` | +| `--builder-follow-redirects` | off | Follow HTTP redirects from the builder (up to ten). Off, a redirect fails the call with an error naming its target, so registrations and signed blocks go only where `--builder` points. 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 12fa2be26..7dc72dc7c 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -474,8 +474,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`, `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 | | +| `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,redirected,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 (`duplicate`: superseded by a newer registration of the same validator in the request) | On every `register_validator` request | outcome=forwarded,failed,dropped,duplicate | | ### Beacon Committee Cache diff --git a/docs/rpc.md b/docs/rpc.md index e63a27332..4af1a220e 100644 --- a/docs/rpc.md +++ b/docs/rpc.md @@ -418,9 +418,19 @@ the chain actor writes, so no request waits on the actor. 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. + locally still pays the registered address. As in Lighthouse, that happens + before the builder is asked: it holds whatever the builder answers, and with + no `--builder` at all. Signatures are not checked here (Lighthouse does not + check them either); the builder does. A validator registered more than once + in one request keeps only its newest registration by `timestamp` (the later + one on a tie, as builders order them), counted as `duplicate`. 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, an empty list included, which is answered without reading + the head state. Future improvement: every other call walks the whole + registry once, with nothing bounding how many walks run at once; a + pubkey-to-index map beside the head state would make it a lookup per + registration. - **`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 f33971d63888b5c813a49c0cd61c5c097d573c50 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Thu, 5 Nov 2026 18:43:07 -0300 Subject: [PATCH 8/8] fix(rpc): tighten the builder client's redirects and bound registry walks Address the bot reviews of the register_validator change: - A redirect's Location is resolved against the URL that answered it and shown without user or password. A scheme-relative target such as `//user:pass@host/x` used to fail to parse and be shown raw, credentials included; one that does not resolve, or resolves to a URL with no host, is now a fixed placeholder, never the raw value. - --builder-follow-redirects follows only 307 and 308, up to ten hops, never from https to http. reqwest's default policy also followed 301, 302 and 303, which resend a POST as a body-less GET, and followed downgrades. A redirect not followed is stopped rather than errored, so it still surfaces as BuilderError::Redirected naming its target. - Only 301, 302, 303, 307 and 308 count as redirects; any other 3xx (304, 300) is an ordinary unexpected status. - register_validator's registry walks take a permit from a two-permit semaphore before spawn_blocking, so concurrent requests queue as suspended tasks instead of filling the blocking pool. The permit moves into the walk, so a client hanging up does not release it early. - BuilderClient's Debug is written out and strips --builder credentials. - Comment why an all-dropped request is a 500 without a builder and a 200 with one. Claude-Session: https://claude.ai/code/session_011vghNneoZ1Ta9W6JNqzjxm --- bin/ethlambda/src/cli.rs | 10 +- .../rpc/src/beacon/builder_client/error.rs | 78 ++++-- .../src/beacon/builder_client/mock_relay.rs | 43 +++- .../net/rpc/src/beacon/builder_client/mod.rs | 81 +++++- .../rpc/src/beacon/builder_client/tests.rs | 238 +++++++++++++++++- crates/net/rpc/src/beacon/registrations.rs | 98 +++++++- docs/cli.md | 2 +- 7 files changed, 486 insertions(+), 64 deletions(-) diff --git a/bin/ethlambda/src/cli.rs b/bin/ethlambda/src/cli.rs index 27bb92bee..249c9ba6c 100644 --- a/bin/ethlambda/src/cli.rs +++ b/bin/ethlambda/src/cli.rs @@ -316,9 +316,13 @@ pub(crate) struct MainnetOptions { #[arg(long, value_parser = parse_builder_url)] pub(crate) builder: Option, - /// Follow HTTP redirects from the builder. Off by default, so registrations - /// and signed blocks go only to the URL `--builder` names: a redirect fails - /// the call, with an error naming where it pointed. + /// Follow the builder's 307 and 308 redirects, up to ten per call and + /// never from https to http. A 301, 302 or 303 is not followed even then: + /// following one resends a POST as a GET without its body. + /// + /// Off by default, so registrations and signed blocks go only to the URL + /// `--builder` names. A redirect not followed fails the call, with an + /// error naming where it pointed. #[arg(long, requires = "builder")] pub(crate) builder_follow_redirects: bool, diff --git a/crates/net/rpc/src/beacon/builder_client/error.rs b/crates/net/rpc/src/beacon/builder_client/error.rs index 3d4463c93..18918dbed 100644 --- a/crates/net/rpc/src/beacon/builder_client/error.rs +++ b/crates/net/rpc/src/beacon/builder_client/error.rs @@ -2,8 +2,8 @@ use std::time::Duration; -use reqwest::StatusCode; use reqwest::header; +use reqwest::{StatusCode, Url}; use super::{read_body, truncate_error_text}; @@ -27,10 +27,13 @@ pub enum BuilderError { /// A status the call does not expect. #[error("builder answered {status}: {body}")] Status { status: u16, body: String }, - /// A redirect, which the client does not follow unless told to. + /// A redirect the client did not follow: any redirect without + /// `--builder-follow-redirects`, and with it a 301, 302 or 303, an https + /// to http downgrade, or the hop past the tenth. #[error( - "builder redirected the request to {location} ({status}); redirects are not \ - followed unless --builder-follow-redirects is set" + "builder redirected the request to {location} ({status}), which was not followed; \ + only --builder-follow-redirects follows a redirect, and then only a 307 or 308, \ + up to ten, never from https to http" )] Redirected { status: u16, location: String }, /// A body that does not decode as what the call returns. @@ -70,7 +73,7 @@ impl BuilderError { /// A redirect carries its target. pub(super) async fn from_status(response: reqwest::Response, deadline: Duration) -> Self { let status = response.status(); - if status.is_redirection() { + if is_redirect(status) { return Self::Redirected { status: status.as_u16(), location: shown_location(&response), @@ -106,25 +109,60 @@ impl BuilderError { } } -/// A redirect's target as an error shows it: without any user or password, -/// since a URL may carry credentials (see `from_reqwest`), and cut to the -/// error text cap. +/// Whether `status` is a redirect: one of the five statuses a client +/// follows, and the only ones reqwest consults its redirect policy on. Any +/// other 3xx (a 304 answering a conditional request, a 300 offering choices) +/// sends the client nowhere, so it is an ordinary unexpected status. +fn is_redirect(status: StatusCode) -> bool { + matches!( + status, + StatusCode::MOVED_PERMANENTLY + | StatusCode::FOUND + | StatusCode::SEE_OTHER + | StatusCode::TEMPORARY_REDIRECT + | StatusCode::PERMANENT_REDIRECT + ) +} + +/// What a redirect shows when its target cannot be shown safely. +const UNSHOWN_LOCATION: &str = "a Location that does not resolve to a URL with a host"; + +/// A redirect's target as an error shows it. fn shown_location(response: &reqwest::Response) -> String { let Some(value) = response.headers().get(header::LOCATION) else { return "nowhere (no Location header)".to_owned(); }; - let location = String::from_utf8_lossy(value.as_bytes()).into_owned(); - let location = match reqwest::Url::parse(&location) { - Ok(mut url) if !url.username().is_empty() || url.password().is_some() => { - // Neither fails on a URL that already has a user or password. - let _ = url.set_username(""); - let _ = url.set_password(None); - url.to_string() - } - // A relative target, or one with nothing to hide, is shown as sent. - _ => location, - }; - truncate_error_text(location) + shown_target(response.url(), value.as_bytes()) +} + +/// `location` resolved against `base`, the URL that answered with it, as a +/// client following it would, then shown without any user or password and +/// cut to the error text cap: the error ends up in logs and response bodies +/// (see `from_reqwest`). +/// +/// Never the raw value. `//user:pass@host/x` is a relative reference, so it +/// does not parse as a URL on its own, yet it carries credentials; resolved, +/// it is an absolute URL whose credentials can be removed. Resolving also +/// strips any `--builder` credentials a relative target would inherit from +/// `base`. A value that does not resolve, or resolves to a URL with no host +/// to hold credentials (`mailto:`), is shown as a fixed placeholder. +pub(super) fn shown_target(base: &Url, location: &[u8]) -> String { + let location = String::from_utf8_lossy(location); + base.join(&location) + .ok() + .and_then(without_credentials) + .map_or_else( + || UNSHOWN_LOCATION.to_owned(), + |url| truncate_error_text(url.into()), + ) +} + +/// `url` without its user and password, or `None` when it cannot hold them +/// and so cannot have them removed either. +pub(super) fn without_credentials(mut url: Url) -> Option { + url.set_username("").ok()?; + url.set_password(None).ok()?; + Some(url) } /// builder-specs' `ErrorMessage`, of which only the message is kept. 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 f8f1e16de..7a614cdd0 100644 --- a/crates/net/rpc/src/beacon/builder_client/mock_relay.rs +++ b/crates/net/rpc/src/beacon/builder_client/mock_relay.rs @@ -7,7 +7,8 @@ //! 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`, and answers -//! every request with a 307 after `redirect_to`. +//! every request with a 307 after `redirect_to`, or another status after +//! `redirect_with`. use std::sync::{Arc, Mutex}; use std::time::Duration; @@ -58,9 +59,11 @@ struct RelayState { unavailable: bool, /// `registerValidator` calls received, whatever their outcome. registration_calls: usize, - /// Where every request is sent instead, with a 307: this base URL - /// followed by the request's own path. - redirect_to: Option, + /// Where every request is sent instead, and with what status: this base + /// URL followed by the request's own path. + redirect_to: Option<(StatusCode, String)>, + /// Redirects answered, whether or not the client followed them. + redirects_served: usize, } /// A fixed answer to every `getHeader`. @@ -169,7 +172,20 @@ impl MockRelay { /// own path, as a builder that moved does. A 307 keeps the method and the /// body, so a client that follows it lands a POST on `base` intact. pub(crate) fn redirect_to(&self, base: &str) { - self.state.lock().unwrap().redirect_to = Some(base.to_owned()); + self.redirect_with(StatusCode::TEMPORARY_REDIRECT, base); + } + + /// [`redirect_to`](Self::redirect_to), but answering with `status`. A + /// 301, 302 or 303 has a client that follows it resend a POST as a GET + /// without its body. `base` is sent as given, so it may be relative or + /// malformed. + pub(crate) fn redirect_with(&self, status: StatusCode, base: &str) { + self.state.lock().unwrap().redirect_to = Some((status, base.to_owned())); + } + + /// How many requests were answered with a redirect. + pub(crate) fn redirects_served(&self) -> usize { + self.state.lock().unwrap().redirects_served } pub(crate) fn blinded_blocks(&self) -> Vec { @@ -213,16 +229,19 @@ impl MockRelay { } async fn redirect(State(state): State, request: Request, next: Next) -> Response { - let target = state.lock().unwrap().redirect_to.clone(); - let Some(base) = target else { + let target = { + let mut state = state.lock().unwrap(); + let target = state.redirect_to.clone(); + if target.is_some() { + state.redirects_served += 1; + } + target + }; + let Some((status, base)) = target else { return next.run(request).await; }; let location = format!("{base}{}", request.uri().path()); - ( - StatusCode::TEMPORARY_REDIRECT, - [(header::LOCATION, location)], - ) - .into_response() + (status, [(header::LOCATION, location)]).into_response() } async fn header( diff --git a/crates/net/rpc/src/beacon/builder_client/mod.rs b/crates/net/rpc/src/beacon/builder_client/mod.rs index d3d00f637..8017c6a61 100644 --- a/crates/net/rpc/src/beacon/builder_client/mod.rs +++ b/crates/net/rpc/src/beacon/builder_client/mod.rs @@ -16,13 +16,16 @@ //! //! Redirects are not followed unless `--builder-follow-redirects` asks for //! it, so the node sends registrations and signed blocks only to the URL the -//! operator named. A refused redirect is an error naming its target. +//! operator named. Even then only a 307 or 308 is followed, up to ten hops +//! and never from https to http (see `follows`). A redirect not followed is +//! an error naming its target. //! //! `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. +use std::fmt; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; @@ -33,9 +36,9 @@ use ethlambda_types::beacon::builder::{ use ethlambda_types::beacon::fork::ForkName; use ethlambda_types::beacon::primitives::{BlsPubkey, ExecutionBlockHash, Slot}; use libssz::{SszDecode as _, SszEncode as _}; -use reqwest::StatusCode; use reqwest::header::{self, HeaderMap}; use reqwest::redirect::Policy; +use reqwest::{StatusCode, Url}; use tracing::info; use crate::metrics::observe_builder_request; @@ -73,8 +76,16 @@ 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); +/// The most redirects one call follows with `--builder-follow-redirects`, +/// which is reqwest's own default cap. +const MAX_REDIRECTS: usize = 10; + /// A builder API endpoint: a relay, or MEV-Boost in front of several. -#[derive(Debug, Clone)] +/// +/// `Debug` is written out rather than derived: `base_url` may carry basic +/// auth credentials from `--builder`, and a client logged with `{:?}` must +/// not print them. +#[derive(Clone)] pub struct BuilderClient { base_url: String, http: reqwest::Client, @@ -106,13 +117,14 @@ impl BuilderClient { Self::with_redirects(base_url, header_timeout, user_agent, false) } - /// [`new`](Self::new), but following up to ten redirects (reqwest's - /// default) when `follow_redirects` is set. + /// [`new`](Self::new), but following a 307 or 308 redirect, up to ten + /// of them and never from https to http, when `follow_redirects` is set + /// (see `follows`). /// /// Without it every call goes exactly where `base_url` points. A redirect - /// fails the call as [`BuilderError::Redirected`], which names the target, - /// rather than carrying registrations or a signed block somewhere - /// `--builder` does not name. + /// not followed fails the call as [`BuilderError::Redirected`], which + /// names the target, rather than carrying registrations or a signed block + /// somewhere `--builder` does not name. pub fn with_redirects( base_url: impl Into, header_timeout: Duration, @@ -122,7 +134,17 @@ impl BuilderClient { let base_url = base_url.into(); Self::check_url(&base_url)?; let redirects = if follow_redirects { - Policy::default() + // `stop`, never `error`: a stopped redirect comes back as the 3xx + // response itself, which `from_status` turns into `Redirected` + // naming its target, where an error would surface as a bare + // transport failure. + Policy::custom(|attempt| { + if follows(attempt.status(), attempt.url(), attempt.previous()) { + attempt.follow() + } else { + attempt.stop() + } + }) } else { Policy::none() }; @@ -349,6 +371,47 @@ impl BuilderClient { } } +impl fmt::Debug for BuilderClient { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let base_url = Url::parse(&self.base_url) + .ok() + .and_then(error::without_credentials) + .map_or_else(|| "".to_owned(), String::from); + f.debug_struct("BuilderClient") + .field("base_url", &base_url) + .field("header_timeout", &self.header_timeout) + .field("json_only", &self.json_only.load(Ordering::Relaxed)) + .finish_non_exhaustive() + } +} + +/// Whether `--builder-follow-redirects` follows a redirect answered with +/// `status` to `next`, after the chain of URLs `previous` (the first of which +/// is the URL the call started at). +/// +/// - Only a 307 or 308. reqwest turns a POST answered with a 301, 302 or 303 +/// into a GET without its body before asking, and does not say which +/// method the request had, so following one would send `registerValidator` +/// or `submitBlindedBlockV2` on as an empty GET. Those three are not +/// followed for any call, GETs included. +/// - Never from https to http, which would send registrations or a signed +/// block in the clear after the operator chose TLS. +/// - Only to http or https, the only schemes reqwest follows: it fails a call +/// told to follow any other as a transport error, which would not name the +/// target. +/// - At most [`MAX_REDIRECTS`] hops, as reqwest's default policy: a custom +/// policy catches no loop on its own. +fn follows(status: StatusCode, next: &Url, previous: &[Url]) -> bool { + let keeps_method = matches!( + status, + StatusCode::TEMPORARY_REDIRECT | StatusCode::PERMANENT_REDIRECT + ); + let downgrade = + next.scheme() == "http" && previous.last().is_some_and(|url| url.scheme() == "https"); + let followable = matches!(next.scheme(), "http" | "https"); + keeps_method && !downgrade && followable && previous.len() <= MAX_REDIRECTS +} + /// A bid, decoded by the response's own `Content-Type`. async fn decode_bid( response: reqwest::Response, diff --git a/crates/net/rpc/src/beacon/builder_client/tests.rs b/crates/net/rpc/src/beacon/builder_client/tests.rs index 9cbe671d2..24fbae7c3 100644 --- a/crates/net/rpc/src/beacon/builder_client/tests.rs +++ b/crates/net/rpc/src/beacon/builder_client/tests.rs @@ -9,9 +9,11 @@ use ethlambda_types::beacon::fork::ForkName; use ethlambda_types::beacon::primitives::{ BlsPubkey, BlsSignature, ExecutionBlockHash, Root, Uint256, Version, }; +use reqwest::Url; +use super::error::shown_target; use super::mock_relay::{MockRelay, signed_registration, unreachable_url}; -use super::{BuilderClient, BuilderError, VersionedBid}; +use super::{BuilderClient, BuilderError, VersionedBid, follows}; /// Any version will do: the relay and the signers only have to agree on it. const GENESIS_FORK_VERSION: Version = [0x10, 0x00, 0x00, 0x38]; @@ -200,20 +202,135 @@ async fn a_redirect_is_refused_and_names_its_target() { assert_eq!(target.registration_calls(), 0); } +fn following_client_for(relay: &MockRelay) -> BuilderClient { + BuilderClient::with_redirects(relay.url.clone(), HEADER_TIMEOUT, "ethlambda/test", true) + .unwrap() +} + #[tokio::test] -async fn a_redirect_is_followed_when_asked() { - let moved = MockRelay::start(GENESIS_FORK_VERSION).await; - let target = MockRelay::start(GENESIS_FORK_VERSION).await; - moved.redirect_to(&target.url); - let client = - BuilderClient::with_redirects(moved.url.clone(), HEADER_TIMEOUT, "ethlambda/test", true) +async fn a_307_or_308_is_followed_when_asked() { + for status in [ + StatusCode::TEMPORARY_REDIRECT, + StatusCode::PERMANENT_REDIRECT, + ] { + let moved = MockRelay::start(GENESIS_FORK_VERSION).await; + let target = MockRelay::start(GENESIS_FORK_VERSION).await; + moved.redirect_with(status, &target.url); + let registrations = vec![signed_registration(0, GENESIS_FORK_VERSION)]; + + following_client_for(&moved) + .register_validators(®istrations) + .await .unwrap(); - let registrations = vec![signed_registration(0, GENESIS_FORK_VERSION)]; - client.register_validators(®istrations).await.unwrap(); + // Both keep the POST and its body, so the target got them whole. + assert_eq!(target.registrations(), registrations, "{status}"); + } +} + +#[tokio::test] +async fn a_301_302_or_303_is_not_followed_even_when_asked() { + for status in [ + StatusCode::MOVED_PERMANENTLY, + StatusCode::FOUND, + StatusCode::SEE_OTHER, + ] { + let moved = MockRelay::start(GENESIS_FORK_VERSION).await; + let target = MockRelay::start(GENESIS_FORK_VERSION).await; + moved.redirect_with(status, &target.url); + let registrations = vec![signed_registration(0, GENESIS_FORK_VERSION)]; + + let err = following_client_for(&moved) + .register_validators(®istrations) + .await + .unwrap_err(); + + // Followed, the POST would have reached the target as an empty GET, + // and the call would have failed on the target's 405 instead. + let answered = status.as_u16(); + assert!( + matches!(err, BuilderError::Redirected { status, .. } if status == answered), + "{err:?}" + ); + assert_eq!(target.registration_calls(), 0, "{status}"); + } +} + +#[tokio::test] +async fn a_redirect_loop_stops_after_ten_hops() { + let looping = MockRelay::start(GENESIS_FORK_VERSION).await; + looping.redirect_to(&looping.url); + + let err = following_client_for(&looping).status().await.unwrap_err(); + + // The first request and ten followed redirects; the eleventh redirect is + // the answer, named rather than failed as a transport error. + assert!( + matches!(err, BuilderError::Redirected { status: 307, .. }), + "{err:?}" + ); + assert_eq!(looping.redirects_served(), 11); +} + +fn url(value: &str) -> Url { + Url::parse(value).unwrap() +} + +#[test] +fn only_a_307_or_308_is_followed() { + let next = url("https://relay.example/eth/v1/builder/validators"); + let previous = [url("https://mev-boost.example/eth/v1/builder/validators")]; + for status in [ + StatusCode::TEMPORARY_REDIRECT, + StatusCode::PERMANENT_REDIRECT, + ] { + assert!(follows(status, &next, &previous), "{status}"); + } + for status in [ + StatusCode::MOVED_PERMANENTLY, + StatusCode::FOUND, + StatusCode::SEE_OTHER, + ] { + assert!(!follows(status, &next, &previous), "{status}"); + } +} + +#[test] +fn a_downgrade_from_https_to_http_is_refused() { + // The mock relay serves plain http, so the downgrade is checked here + // rather than end to end. + let status = StatusCode::TEMPORARY_REDIRECT; + let https = url("https://relay.example/eth/v1/builder/status"); + let http = url("http://relay.example/eth/v1/builder/status"); + + assert!(!follows(status, &http, std::slice::from_ref(&https))); + // Whatever came before: one https hop is enough to rule out http after it. + assert!(!follows(status, &http, &[http.clone(), https.clone()])); + // An upgrade, or staying on one scheme, is fine. + assert!(follows(status, &https, std::slice::from_ref(&http))); + assert!(follows(status, &https, std::slice::from_ref(&https))); + assert!(follows(status, &http, std::slice::from_ref(&http))); +} - // A 307 keeps the POST and its body, so the target got them whole. - assert_eq!(target.registrations(), registrations); +#[test] +fn a_redirect_to_a_scheme_other_than_http_is_not_followed() { + let previous = [url("http://relay.example/")]; + let ftp = url("ftp://relay.example/eth/v1/builder/status"); + assert!(!follows(StatusCode::TEMPORARY_REDIRECT, &ftp, &previous)); +} + +#[test] +fn at_most_ten_redirects_are_followed() { + let status = StatusCode::TEMPORARY_REDIRECT; + let next = url("http://relay.example/next"); + // `previous` starts with the URL the call was sent to, so ten URLs in it + // means nine redirects followed so far, and this one is the tenth. + let mut previous: Vec = (0..10) + .map(|hop| url(&format!("http://relay.example/{hop}"))) + .collect(); + assert!(follows(status, &next, &previous)); + previous.push(url("http://relay.example/10")); + assert!(!follows(status, &next, &previous)); } #[tokio::test] @@ -232,6 +349,105 @@ async fn a_redirect_target_is_shown_without_its_credentials() { assert!(!message.contains("operator"), "{message}"); } +#[tokio::test] +async fn a_scheme_relative_target_is_shown_absolute_without_its_credentials() { + // `//user:pass@host/x` does not parse as a URL on its own; it is only + // stripped once resolved against the URL that answered. + let moved = MockRelay::start(GENESIS_FORK_VERSION).await; + moved.redirect_to("//operator:hunter2@127.0.0.1:9"); + + let err = client_for(&moved).status().await.unwrap_err(); + + let message = err.to_string(); + assert!( + message.contains("http://127.0.0.1:9/eth/v1/builder/status"), + "{message}" + ); + assert!(!message.contains("hunter2"), "{message}"); + assert!(!message.contains("operator"), "{message}"); +} + +#[tokio::test] +async fn a_relative_target_is_shown_resolved_against_the_builder() { + let moved = MockRelay::start(GENESIS_FORK_VERSION).await; + moved.redirect_to("/moved"); + + let err = client_for(&moved).status().await.unwrap_err(); + + let location = format!("{}/moved/eth/v1/builder/status", moved.url); + assert!( + matches!(&err, BuilderError::Redirected { location: shown, .. } if *shown == location), + "{err:?}" + ); +} + +#[tokio::test] +async fn an_unparsable_target_is_not_shown() { + // An unterminated IPv6 host: not a URL, resolved or not. + let moved = MockRelay::start(GENESIS_FORK_VERSION).await; + moved.redirect_to("http://operator:hunter2@[::1"); + + let err = client_for(&moved).status().await.unwrap_err(); + + let BuilderError::Redirected { location, .. } = &err else { + panic!("expected a redirect, got {err:?}"); + }; + assert_eq!( + location, + "a Location that does not resolve to a URL with a host" + ); + let message = err.to_string(); + assert!(!message.contains("hunter2"), "{message}"); + assert!(!message.contains("operator"), "{message}"); +} + +#[test] +fn a_relative_target_does_not_inherit_the_builders_credentials() { + let base = url("http://operator:hunter2@relay.example/eth/v1/builder/status"); + assert_eq!(shown_target(&base, b"/moved"), "http://relay.example/moved"); +} + +#[test] +fn a_target_with_no_host_is_not_shown() { + // It resolves, but to a URL with no host, so it has no user or password + // to remove and is not shown at all. + let base = url("http://relay.example/eth/v1/builder/status"); + assert_eq!( + shown_target(&base, b"mailto:operator:hunter2@relay.example"), + "a Location that does not resolve to a URL with a host" + ); +} + +#[tokio::test] +async fn a_3xx_that_is_not_a_redirect_is_an_unexpected_status() { + for status in [StatusCode::MULTIPLE_CHOICES, StatusCode::NOT_MODIFIED] { + let relay = MockRelay::start(GENESIS_FORK_VERSION).await; + relay.answer_headers_with(status, TEXT, Vec::new(), None); + let answered = status.as_u16(); + let result = get_header(&client_for(&relay)).await; + assert!( + matches!(&result, Err(BuilderError::Status { status, .. }) if *status == answered), + "{result:?}" + ); + } +} + +#[test] +fn debug_does_not_show_the_urls_credentials() { + let client = BuilderClient::new( + "http://operator:hunter2@127.0.0.1:18550", + HEADER_TIMEOUT, + "ethlambda/test", + ) + .unwrap(); + + let shown = format!("{client:?}"); + + assert!(shown.contains("http://127.0.0.1:18550/"), "{shown}"); + assert!(!shown.contains("hunter2"), "{shown}"); + assert!(!shown.contains("operator"), "{shown}"); +} + const JSON: &str = "application/json"; const TEXT: &str = "text/plain"; const SSZ: &str = "application/octet-stream"; diff --git a/crates/net/rpc/src/beacon/registrations.rs b/crates/net/rpc/src/beacon/registrations.rs index 3ec1e07bc..da64c4112 100644 --- a/crates/net/rpc/src/beacon/registrations.rs +++ b/crates/net/rpc/src/beacon/registrations.rs @@ -19,6 +19,7 @@ //! the head state. use std::collections::{HashMap, HashSet}; +use std::sync::Arc; use axum::body::Bytes; use axum::extract::{DefaultBodyLimit, State}; @@ -34,6 +35,7 @@ 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 tokio::sync::Semaphore; use tracing::warn; use super::builder_client::{BuilderClient, BuilderError}; @@ -41,23 +43,53 @@ use super::validator::{FeeRecipients, head}; use super::{ApiError, BodyEncoding}; use crate::metrics::inc_builder_registrations; +/// How many registry walks (`active_or_pending`) may run at once. +const CONCURRENT_REGISTRY_SCANS: usize = 2; + +/// The permits a registry walk takes, shared by every request this router +/// serves. +/// +/// Each non-empty request walks the whole registry on the blocking pool, +/// which the duty endpoints share. Unbounded, a burst of requests (every +/// validator client of a large operator re-registering at once, or a client +/// that retries hard) would take every blocking thread for as long as the +/// walks last. With the bound, a request beyond it waits for a permit as a +/// suspended task, holding no thread. The walk itself is what a pubkey index +/// would remove (see `active_or_pending`). +#[derive(Clone)] +struct RegistryScans(Arc); + +impl Default for RegistryScans { + fn default() -> Self { + Self(Arc::new(Semaphore::new(CONCURRENT_REGISTRY_SCANS))) + } +} + pub(crate) fn routes() -> Router { + routes_with(RegistryScans::default()) +} + +fn routes_with(scans: RegistryScans) -> 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)), - ) + Router::new() + .route( + "/eth/v1/validator/register_validator", + post(post_register_validator) + .layer(DefaultBodyLimit::max(super::MAX_PUBLISH_BODY_BYTES)), + ) + .layer(Extension(scans)) } async fn post_register_validator( State(store): State, Extension(builder): Extension>, Extension(fee_recipients): Extension, + Extension(scans): Extension, headers: HeaderMap, body: Bytes, ) -> Response { - register(store, builder, fee_recipients, &headers, &body) + register(store, builder, fee_recipients, scans, &headers, &body) .await .map_or_else(IntoResponse::into_response, |()| { axum::http::StatusCode::OK.into_response() @@ -71,6 +103,7 @@ async fn register( store: Store, builder: Option, fee_recipients: FeeRecipients, + scans: RegistryScans, headers: &HeaderMap, body: &[u8], ) -> Result<(), ApiError> { @@ -90,9 +123,18 @@ async fn register( inc_builder_registrations("duplicate", (submitted - registrations.len()) as u64); // The head-state read and the walk over the whole registry are blocking - // work, as the duty endpoints' walks are. + // work, as the duty endpoints' walks are, and take a permit first (see + // `RegistryScans`). The permit moves into the walk, so it is released + // when the walk ends: a validator client that hangs up drops this + // handler, not the walk already running for it. + let permit = scans + .0 + .acquire_owned() + .await + .map_err(|_| ApiError::Internal("checking the registrations failed"))?; let candidates = registrations.len(); let kept = tokio::task::spawn_blocking(move || { + let _permit = permit; 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. @@ -112,6 +154,11 @@ async fn register( recipients.insert(*index, registration.message.fee_recipient); } } + // The builder check comes before the empty one on purpose, so a request + // whose every validator was dropped is a 500 without a builder and a 200 + // with one: the 500 is how a validator client configured for a builder + // learns there is none, whatever its body held, while a builder that is + // there has simply been sent nothing it would take. let builder = builder.ok_or(ApiError::Internal(NO_BUILDER))?; if kept.is_empty() { return Ok(()); @@ -195,8 +242,9 @@ fn newest_per_validator( /// index. One pass over the registry, whatever the number of registrations. /// /// Future improvement: that pass happens on every call, over the whole -/// registry, with nothing bounding how many run at once. A pubkey-to-index -/// map kept beside the head state would make it one lookup per registration. +/// registry. `RegistryScans` bounds how many run at once, but each still +/// costs a full walk; a pubkey-to-index map kept beside the head state would +/// make it one lookup per registration, as Lighthouse's pubkey cache does. fn active_or_pending( state: &BeaconState, registrations: Vec, @@ -663,6 +711,40 @@ mod tests { assert_eq!(relay.registration_calls(), 0); } + #[tokio::test] + async fn a_walk_beyond_the_bound_waits_for_a_permit() { + let store = store_with_exited(&[]); + let relay = MockRelay::start(genesis_fork_version(&store)).await; + let registration = signed_registration(0, genesis_fork_version(&store)); + let scans = RegistryScans::default(); + let app = routes_with(scans.clone()) + .with_state(store) + .layer(Extension(client_for(&relay))) + .layer(Extension(FeeRecipients::default())); + let every_permit = CONCURRENT_REGISTRY_SCANS as u32; + let held = scans + .0 + .clone() + .acquire_many_owned(every_permit) + .await + .unwrap(); + + // An empty request walks nothing, so it needs no permit. + let response = app.clone().oneshot(json_request(&[])).await.unwrap(); + assert_eq!(response.status(), StatusCode::OK); + + let request = json_request(std::slice::from_ref(®istration)); + let waiting = tokio::spawn(app.oneshot(request)); + tokio::time::sleep(Duration::from_millis(100)).await; + assert!(!waiting.is_finished(), "ran without a permit"); + assert_eq!(relay.registration_calls(), 0); + + drop(held); + let response = waiting.await.unwrap().unwrap(); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(relay.registrations(), [registration]); + } + #[tokio::test] async fn an_empty_list_without_a_builder_is_a_server_error() { let store = store_with_exited(&[]); diff --git a/docs/cli.md b/docs/cli.md index 38351f903..8125a8b40 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -116,7 +116,7 @@ from a normal build. | `--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` | -| `--builder-follow-redirects` | off | Follow HTTP redirects from the builder (up to ten). Off, a redirect fails the call with an error naming its target, so registrations and signed blocks go only where `--builder` points. Requires `--builder` | +| `--builder-follow-redirects` | off | Follow the builder's 307 and 308 redirects, up to ten per call, never from https to http. A 301, 302 or 303 is not followed even then, since following one resends a POST as a GET without its body. Off, every redirect fails the call with an error naming its target, so registrations and signed blocks go only where `--builder` points; on, so does any redirect it does not follow. 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