feat(beacon): --builder and register_validator

bin/ethlambda/src/cli.rs

@@ -307,6 +307,24 @@ pub(crate) struct MainnetOptions {
     /// `eth-clients` publishes and kurtosis mounts at `/network-configs`.
     #[arg(long, default_value = crate::network::DEFAULT_NETWORK)]
     pub(crate) network: String,
+
+    /// Base URL of a builder API endpoint, normally MEV-Boost (for example
+    /// `http://127.0.0.1:18550`) or a relay directly. Without it every block
+    /// is built locally, and `register_validator` answers 500 so a validator
+    /// client configured for a builder hears that it has none.
+    #[arg(long)]
+    pub(crate) builder: Option<String>,
+
+    /// How long block production waits for the builder's bid, in
+    /// milliseconds. Also sent to the builder as `X-Timeout-Ms`, so it can
+    /// hold its answer for as long as this node will wait.
+    #[arg(
+        long,
+        default_value_t = 1000,
+        requires = "builder",
+        value_parser = clap::value_parser!(u64).range(1..),
+    )]
+    pub(crate) builder_header_timeout: u64,
 }
 
 /// Flags only the lean chain takes.
@@ -1020,6 +1038,56 @@ mod tests {
         );
     }
 
+    #[test]
+    fn the_builder_is_off_by_default() {
+        let beacon = parse_beacon(beacon_args());
+
+        assert_eq!(beacon.mainnet.builder, None);
+        assert_eq!(beacon.mainnet.builder_header_timeout, 1000);
+    }
+
+    #[test]
+    fn the_builder_flags_parse() {
+        let mut args = beacon_args();
+        args.extend([
+            "--builder",
+            "http://127.0.0.1:18550",
+            "--builder-header-timeout",
+            "750",
+        ]);
+
+        let options: Options = parse_beacon(args).into();
+
+        match options.network {
+            Network::Mainnet { mainnet, .. } => {
+                assert_eq!(mainnet.builder.as_deref(), Some("http://127.0.0.1:18550"));
+                assert_eq!(mainnet.builder_header_timeout, 750);
+            }
+            other => panic!("expected the mainnet network, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn a_zero_header_timeout_is_refused() {
+        let mut args = beacon_args();
+        args.extend([
+            "--builder",
+            "http://127.0.0.1:18550",
+            "--builder-header-timeout",
+            "0",
+        ]);
+
+        try_parse_from(args).expect_err("a zero timeout must not parse");
+    }
+
+    #[test]
+    fn a_header_timeout_needs_a_builder() {
+        let mut args = beacon_args();
+        args.extend(["--builder-header-timeout", "750"]);
+
+        try_parse_from(args).expect_err("a timeout without a builder must not parse");
+    }
+
     #[test]
     fn both_execution_flags_together_pair_into_one_value() {
         let mut args = beacon_args();

bin/ethlambda/src/main.rs

@@ -414,6 +414,35 @@ async fn run_node(options: Options) -> eyre::Result<()> {
         Network::Mainnet { .. } => false,
     });
 
+    // A malformed URL fails here, at startup, rather than at the first
+    // proposal; an unreachable builder only warns, since it may come up
+    // after this node does.
+    let builder = match &network {
+        Network::Lean(_) => None,
+        Network::Mainnet { mainnet, .. } => mainnet
+            .builder
+            .as_deref()
+            .map(|url| {
+                ethlambda_rpc::BuilderClient::new(
+                    url,
+                    std::time::Duration::from_millis(mainnet.builder_header_timeout),
+                    version::client_version(),
+                )
+            })
+            .transpose()?,
+    };
+    if let Some(builder) = builder.clone() {
+        tokio::spawn(async move {
+            let _ = builder
+                .status()
+                .await
+                .inspect(|()| info!("Builder is up"))
+                .inspect_err(|err| {
+                    warn!(%err, "Builder is not answering; proposals build locally until it does")
+                });
+        });
+    }
+
     // Shared, runtime-readable sync status. The blockchain actor writes it each
     // tick (alongside the `lean_node_sync_status` metric); the RPC
     // `/lean/v0/node/syncing` endpoint reads it. Seeded to Idle, matching the
@@ -805,6 +834,7 @@ async fn run_node(options: Options) -> eyre::Result<()> {
                     events: rpc_events,
                     client_version: version::engine_client_version(),
                     identity: rpc_identity,
+                    builder,
                 },
                 local_peer_id,
                 rpc_shutdown,

crates/net/rpc/src/beacon/builder_client/mock_relay.rs

@@ -6,6 +6,7 @@
 //! registration signatures and refuses SSZ registrations, records blinded
 //! block submissions, and has `answer_headers_with` and
 //! `answer_submissions_with` to answer a call with exactly what a test needs.
+//! It serves `status` with 200, or 503 after `make_unavailable`.
 
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
@@ -51,6 +52,10 @@ struct RelayState {
     submit_status: Option<StatusCode>,
     /// Submissions that arrived SSZ-encoded, refused or not.
     ssz_submissions: usize,
+    /// Whether `status` answers 503.
+    unavailable: bool,
+    /// `registerValidator` calls received, whatever their outcome.
+    registration_calls: usize,
 }
 
 /// A fixed answer to every `getHeader`.
@@ -92,6 +97,7 @@ impl MockRelay {
             )
             .route("/eth/v1/builder/validators", post(register))
             .route("/eth/v2/builder/blinded_blocks", post(submit))
+            .route("/eth/v1/builder/status", get(status))
             .with_state(state.clone());
         let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
         let url = format!("http://{}", listener.local_addr().unwrap());
@@ -148,6 +154,11 @@ impl MockRelay {
         self.state.lock().unwrap().submit_status = Some(status);
     }
 
+    /// Answer `status` with 503, as a builder that is not ready does.
+    pub(crate) fn make_unavailable(&self) {
+        self.state.lock().unwrap().unavailable = true;
+    }
+
     pub(crate) fn blinded_blocks(&self) -> Vec<SignedBlindedBeaconBlock> {
         self.state.lock().unwrap().blinded_blocks.clone()
     }
@@ -156,6 +167,11 @@ impl MockRelay {
         self.state.lock().unwrap().ssz_submissions
     }
 
+    /// How many `registerValidator` calls arrived, whatever their outcome.
+    pub(crate) fn registration_calls(&self) -> usize {
+        self.state.lock().unwrap().registration_calls
+    }
+
     pub(crate) fn registrations(&self) -> Vec<SignedValidatorRegistrationV1> {
         self.state.lock().unwrap().registrations.clone()
     }
@@ -244,6 +260,7 @@ async fn header(
 
 async fn register(State(state): State<Shared>, headers: HeaderMap, body: Bytes) -> Response {
     let mut state = state.lock().unwrap();
+    state.registration_calls += 1;
     // The client always registers in JSON (see `register_validators`); an
     // SSZ registration here is a regression.
     if is_ssz_request(&headers) {
@@ -264,6 +281,14 @@ async fn register(State(state): State<Shared>, headers: HeaderMap, body: Bytes)
     StatusCode::OK.into_response()
 }
 
+async fn status(State(state): State<Shared>) -> Response {
+    if state.lock().unwrap().unavailable {
+        StatusCode::SERVICE_UNAVAILABLE.into_response()
+    } else {
+        StatusCode::OK.into_response()
+    }
+}
+
 async fn submit(State(state): State<Shared>, headers: HeaderMap, body: Bytes) -> Response {
     let mut state = state.lock().unwrap();
     let ssz = is_ssz_request(&headers);

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);

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, .. })
+    ));
+}

crates/net/rpc/src/beacon/mod.rs

@@ -34,6 +34,7 @@ pub(crate) mod pool;
 pub(crate) mod proposal;
 pub(crate) mod proposer_preferences;
 pub(crate) mod ptc;
+pub(crate) mod registrations;
 pub(crate) mod states;
 pub(crate) mod sync_committee;
 pub(crate) mod validator;
@@ -191,6 +192,7 @@ pub(crate) fn routes(version: &'static str, peer_id: String) -> Router<Store> {
         .merge(config::routes())
         .merge(node::routes(version, peer_id))
         .merge(validator::routes())
+        .merge(registrations::routes())
         .merge(pool::routes())
         .merge(operations::routes())
         .merge(proposal::routes())