feat(beacon): add beacon RPC handlers

Author: iamwehi

src/config.rs

@@ -57,11 +57,23 @@ pub fn default_chains() -> Vec<ChainSpec> {
     ]
 }
 
+/// Default beacon light-client upstream. A SINGLE consistent node — Helios's
+/// `light_client/updates` walk breaks if fanned out across backends at different
+/// sync states (the whole reason the `/beacon` split exists). See `SERVER_CONSENSUS.md`.
+pub const DEFAULT_BEACON_LC_UPSTREAM: &str = "https://lodestar-mainnet.chainsafe.io";
+/// Default beacon blocks upstream. Serves the ~750 KB `/eth/v2/beacon/blocks/{slot}`
+/// payload Helios re-fetches every slot without rate-limiting it.
+pub const DEFAULT_BEACON_BLOCKS_UPSTREAM: &str = "https://ethereum-beacon-api.publicnode.com";
+
 pub struct Config {
     pub bind: SocketAddr,
     pub drpc_base: String,
     pub key: Key,
     pub chains: Vec<ChainSpec>,
+    /// Beacon light-client upstream (clean, in-order `light_client/updates`).
+    pub beacon_lc_upstream: String,
+    /// Beacon blocks upstream (un-rate-limited `/eth/v2/beacon/blocks/{slot}`).
+    pub beacon_blocks_upstream: String,
     pub body_limit: usize,
     /// When true, refuse expensive/nonsensical-over-HTTP JSON-RPC methods so a
     /// leaked endpoint can't burn the shared key on archival traces. Off via
@@ -88,6 +100,14 @@ impl Config {
             drpc_base: drpc_base.trim_end_matches('/').to_string(),
             key: load_key()?,
             chains: default_chains(),
+            beacon_lc_upstream: std::env::var("BEACON_LC_UPSTREAM")
+                .unwrap_or_else(|_| DEFAULT_BEACON_LC_UPSTREAM.into())
+                .trim_end_matches('/')
+                .to_string(),
+            beacon_blocks_upstream: std::env::var("BEACON_BLOCKS_UPSTREAM")
+                .unwrap_or_else(|_| DEFAULT_BEACON_BLOCKS_UPSTREAM.into())
+                .trim_end_matches('/')
+                .to_string(),
             body_limit: env_usize("RPC_BODY_LIMIT_BYTES", 1 << 20), // 1 MiB
             method_denylist: std::env::var("RPC_METHOD_FILTER")
                 .map(|v| v != "off")

src/main.rs

@@ -21,9 +21,17 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         .connect_timeout(cfg.connect_timeout)
         .build()?;
 
+    // The beacon proxy reuses the same HTTPS-only client (cloning shares the
+    // connection pool). Its ~20s timeout covers the large ~750 KB block fetch.
+    let beacon = upstream::Beacon::new(
+        client.clone(),
+        cfg.beacon_lc_upstream,
+        cfg.beacon_blocks_upstream,
+    );
     let drpc = upstream::Drpc::new(client, cfg.drpc_base.clone(), cfg.key);
     let state = Arc::new(routes::AppState {
         drpc,
+        beacon,
         chains: cfg.chains,
         method_denylist: cfg.method_denylist,
     });
@@ -39,6 +47,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
             "/v1/{chain}/history/{address}",
             get(routes::history_handler),
         )
+        .route("/beacon/{*path}", get(routes::beacon_handler))
         .layer(DefaultBodyLimit::max(cfg.body_limit))
         .with_state(state);
 

src/routes.rs

@@ -5,6 +5,7 @@
 //!   POST /rpc/{chain}                     JSON-RPC passthrough (alias -> dRPC slug)
 //!   GET  /v1/{chain}/balances/{address}   Wallet-API balances for one chain
 //!   GET  /v1/{chain}/history/{address}    Wallet-API tx history for one chain
+//!   GET  /beacon/{*path}                  Beacon light-client proxy for Helios
 //!
 //! The wallet-API routes are a thin per-chain mirror of dRPC's Wallet API: the
 //! proxy resolves the chain alias to a dRPC slug, injects the shared key,
@@ -15,7 +16,7 @@
 //! source.
 
 use crate::config::ChainSpec;
-use crate::upstream::{Drpc, UpstreamError};
+use crate::upstream::{Beacon, Drpc, UpstreamError};
 use axum::Json;
 use axum::body::Bytes;
 use axum::extract::{Path, RawQuery, State};
@@ -26,6 +27,7 @@ use std::sync::Arc;
 
 pub struct AppState {
     pub drpc: Drpc,
+    pub beacon: Beacon,
     pub chains: Vec<ChainSpec>,
     pub method_denylist: bool,
 }
@@ -50,6 +52,7 @@ pub async fn rpc_handler(
     Path(chain): Path<String>,
     body: Bytes,
 ) -> Result<Response, AppError> {
+    tracing::info!(chain, body_len = body.len(), "TEMP: rpc request received");
     let slug = state.resolve_chain(&chain).ok_or(AppError::UnknownChain)?;
     if state.method_denylist {
         reject_denied_methods(&body)?;
@@ -63,6 +66,7 @@ pub async fn balances_handler(
     Path((chain, address)): Path<(String, String)>,
     RawQuery(query): RawQuery,
 ) -> Result<Response, AppError> {
+    tracing::info!(chain, "TEMP: balances request received");
     let slug = state.resolve_chain(&chain).ok_or(AppError::UnknownChain)?;
     if !is_valid_eth_address(&address) {
         return Err(AppError::BadAddress);
@@ -79,6 +83,7 @@ pub async fn history_handler(
     Path((chain, address)): Path<(String, String)>,
     RawQuery(query): RawQuery,
 ) -> Result<Response, AppError> {
+    tracing::info!(chain, "TEMP: history request received");
     let slug = state.resolve_chain(&chain).ok_or(AppError::UnknownChain)?;
     if !is_valid_eth_address(&address) {
         return Err(AppError::BadAddress);
@@ -90,10 +95,43 @@ pub async fn history_handler(
     Ok(json_passthrough(code, body))
 }
 
+/// Beacon light-client proxy. The wallet's Helios points its consensus RPC at
+/// `{KAO}/beacon` and makes plain HTTP GET / JSON requests. We strip the
+/// `/beacon` prefix, forward the path + query verbatim to the routed upstream,
+/// and return its status, `Content-Type`, and body unchanged.
+///
+/// On a pre-response upstream failure we return a clean 5xx (via `AppError`),
+/// which Helios retries — never a malformed 200 body, which it treats as fatal.
+pub async fn beacon_handler(
+    State(state): State<Arc<AppState>>,
+    Path(path): Path<String>,
+    RawQuery(query): RawQuery,
+) -> Result<Response, AppError> {
+    // Axum's `{*path}` capture drops the leading slash; the upstream wants an
+    // absolute beacon-API path (e.g. `/eth/v2/beacon/blocks/14702657`).
+    let path = format!("/{path}");
+    let (code, content_type, body) = state.beacon.get(&path, query.as_deref()).await?;
+    Ok(beacon_passthrough(code, content_type, body))
+}
+
 /// Build a passthrough response: upstream status (clamped to a valid code),
 /// `application/json`, and the upstream body verbatim.
 fn json_passthrough(code: u16, body: Vec<u8>) -> Response {
-    let status = StatusCode::from_u16(code).unwrap_or(StatusCode::BAD_GATEWAY);
+    // TEMP: a clamp here is the only way this app emits a 502 with a non-JSON
+    // (possibly empty) body — exactly the symptom being chased.
+    let status = StatusCode::from_u16(code).unwrap_or_else(|_| {
+        tracing::warn!(
+            code,
+            body_len = body.len(),
+            "TEMP: invalid upstream status, clamping to 502"
+        );
+        StatusCode::BAD_GATEWAY
+    });
+    tracing::info!(
+        status = code,
+        body_len = body.len(),
+        "TEMP: passthrough response"
+    );
     (
         status,
         [(
@@ -105,6 +143,16 @@ fn json_passthrough(code: u16, body: Vec<u8>) -> Response {
         .into_response()
 }
 
+/// Beacon passthrough: upstream status (clamped) and body verbatim, preserving
+/// the upstream `Content-Type` (Helios parses JSON; we don't rewrite the body).
+fn beacon_passthrough(code: u16, content_type: Option<String>, body: Vec<u8>) -> Response {
+    let status = StatusCode::from_u16(code).unwrap_or(StatusCode::BAD_GATEWAY);
+    let ct = content_type
+        .and_then(|s| HeaderValue::from_str(&s).ok())
+        .unwrap_or_else(|| HeaderValue::from_static("application/json"));
+    (status, [(header::CONTENT_TYPE, ct)], body).into_response()
+}
+
 // ---- validation & method filtering ------------------------------------------
 
 fn is_valid_eth_address(s: &str) -> bool {
@@ -169,6 +217,8 @@ impl IntoResponse for AppError {
             AppError::BadRequest(m) => (StatusCode::BAD_REQUEST, m),
             AppError::Upstream(m) => (StatusCode::BAD_GATEWAY, m),
         };
+        // TEMP:
+        tracing::warn!(status = %status, msg, "TEMP: app error response");
         (
             status,
             Json(serde_json::json!({ "error": { "message": msg } })),

src/upstream.rs

@@ -56,9 +56,24 @@ impl Drpc {
             .body(body.to_vec())
             .send()
             .await
-            .map_err(|_| UpstreamError::Transport)?;
+            .map_err(|e| {
+                // TEMP: `.without_url()` strips the key-bearing URL before logging.
+                tracing::error!(error = %e.without_url(), chain, "TEMP: rpc send failed");
+                UpstreamError::Transport
+            })?;
         let status = resp.status().as_u16();
-        let bytes = resp.bytes().await.map_err(|_| UpstreamError::Transport)?;
+        let bytes = resp.bytes().await.map_err(|e| {
+            // TEMP: `.without_url()` strips the key-bearing URL before logging.
+            tracing::error!(error = %e.without_url(), chain, status, "TEMP: rpc body read failed");
+            UpstreamError::Transport
+        })?;
+        // TEMP:
+        tracing::info!(
+            chain,
+            status,
+            body_len = bytes.len(),
+            "TEMP: rpc upstream response"
+        );
         Ok((status, bytes.to_vec()))
     }
 
@@ -108,13 +123,99 @@ impl Drpc {
             .header(ACCEPT, "application/json")
             .send()
             .await
-            .map_err(|_| UpstreamError::Transport)?;
+            .map_err(|e| {
+                // TEMP: `.without_url()` strips the key-bearing URL before logging.
+                tracing::error!(error = %e.without_url(), "TEMP: get send failed");
+                UpstreamError::Transport
+            })?;
         let status = resp.status().as_u16();
-        let bytes = resp.bytes().await.map_err(|_| UpstreamError::Transport)?;
+        let bytes = resp.bytes().await.map_err(|e| {
+            // TEMP: `.without_url()` strips the key-bearing URL before logging.
+            tracing::error!(error = %e.without_url(), status, "TEMP: get body read failed");
+            UpstreamError::Transport
+        })?;
+        // TEMP:
+        tracing::info!(
+            status,
+            body_len = bytes.len(),
+            "TEMP: get upstream response"
+        );
         Ok((status, bytes.to_vec()))
     }
 }
 
+/// Beacon light-client proxy for the wallet's Helios light client.
+///
+/// Serves the six beacon-API paths under `/beacon` by routing each to whichever
+/// upstream handles *that* request well (see `SERVER_CONSENSUS.md`):
+///  - `/eth/v{1,2}/beacon/blocks/…` → `blocks_upstream` (large payload, must not 429)
+///  - everything else               → `lc_upstream`     (clean, in-order updates)
+///
+/// Like `Drpc`, requests are originated here — no client IP/UA/cookie reaches the
+/// upstream. Once the upstream answers, its status, `Content-Type`, and body are
+/// returned verbatim; only pre-response failures collapse into `Transport` (which
+/// the handler renders as a clean 5xx that Helios retries).
+pub struct Beacon {
+    client: Client,
+    lc_upstream: String,
+    blocks_upstream: String,
+}
+
+impl Beacon {
+    pub fn new(client: Client, lc_upstream: String, blocks_upstream: String) -> Self {
+        Self {
+            client,
+            lc_upstream,
+            blocks_upstream,
+        }
+    }
+
+    /// Pick the upstream for a beacon path. Blocks go to the working-blocks
+    /// upstream; everything else (light_client/*, config/spec, states/*) to the
+    /// clean-updates upstream. `path` is absolute (leading `/`).
+    fn upstream_for(&self, path: &str) -> &str {
+        if path.starts_with("/eth/v1/beacon/blocks/") || path.starts_with("/eth/v2/beacon/blocks/")
+        {
+            &self.blocks_upstream
+        } else {
+            &self.lc_upstream
+        }
+    }
+
+    /// GET passthrough. `path` is the beacon-API path (leading `/`, no `/beacon`
+    /// prefix); `query` is the caller's raw query string, forwarded verbatim.
+    /// Returns the upstream status, `Content-Type`, and body unchanged.
+    pub async fn get(
+        &self,
+        path: &str,
+        query: Option<&str>,
+    ) -> Result<(u16, Option<String>, Vec<u8>), UpstreamError> {
+        let mut url = format!("{}{}", self.upstream_for(path), path);
+        append_query(&mut url, query);
+        let resp = self
+            .client
+            .get(&url)
+            .header(ACCEPT, "application/json")
+            .send()
+            .await
+            .map_err(|e| {
+                tracing::error!(error = %e.without_url(), path, "beacon send failed");
+                UpstreamError::Transport
+            })?;
+        let status = resp.status().as_u16();
+        let content_type = resp
+            .headers()
+            .get(CONTENT_TYPE)
+            .and_then(|v| v.to_str().ok())
+            .map(str::to_owned);
+        let bytes = resp.bytes().await.map_err(|e| {
+            tracing::error!(error = %e.without_url(), path, status, "beacon body read failed");
+            UpstreamError::Transport
+        })?;
+        Ok((status, content_type, bytes.to_vec()))
+    }
+}
+
 /// Append `?query` to `url` when the caller supplied a non-empty query string.
 fn append_query(url: &mut String, query: Option<&str>) {
     if let Some(q) = query