Proxy Module Refactor

Cargo.toml

@@ -23,6 +23,7 @@ reqwest = "0.12.5"
 http = "1.1.0"
 dotenv = "0.15.0"
 tower = "0.4.13"
+tower-http = { version = "0.5.2", features = ["decompression-full"] }
 http-body = "1.0.1"
 http-body-util = "0.1.2"
 hyper = { version = "1.4.1", features = ["full"] }

Dockerfile

@@ -6,7 +6,7 @@
 #
 # Based on https://depot.dev/blog/rust-dockerfile-best-practices
 #
-FROM rust:1.85 AS base
+FROM rust:1.85.1 AS base
 
 ARG FEATURES
 
@@ -61,4 +61,4 @@ WORKDIR /app
 ARG ROLLUP_BOOST_BIN="rollup-boost"
 COPY --from=builder /app/target/release/${ROLLUP_BOOST_BIN} /usr/local/bin/
 
-ENTRYPOINT ["/usr/local/bin/rollup-boost"]
+ENTRYPOINT ["/usr/local/bin/rollup-boost"]

src/auth_layer.rs → src/client/auth.rs

@@ -8,7 +8,7 @@ use std::{
 use tower::{Layer, Service};
 
 /// A layer that adds a new JWT token to every request using `AuthClientService`.
-#[derive(Debug)]
+#[derive(Clone, Debug)]
 pub struct AuthClientLayer {
     secret: JwtSecret,
 }
@@ -44,6 +44,7 @@ impl<S> AuthClientService<S> {
 impl<S, B> Service<http::Request<B>> for AuthClientService<S>
 where
     S: Service<http::Request<B>>,
+    B: std::fmt::Debug,
 {
     type Response = S::Response;
     type Error = S::Error;

src/client/http.rs

@@ -0,0 +1,93 @@
+use crate::server::PayloadSource;
+use alloy_rpc_types_engine::JwtSecret;
+use http::Uri;
+use http_body_util::BodyExt;
+use hyper_rustls::HttpsConnector;
+use hyper_util::client::legacy::Client;
+use hyper_util::client::legacy::connect::HttpConnector;
+use hyper_util::rt::TokioExecutor;
+use jsonrpsee::core::BoxError;
+use jsonrpsee::http_client::HttpBody;
+use opentelemetry::trace::SpanKind;
+use tower::{Service as _, ServiceBuilder, ServiceExt};
+use tower_http::decompression::{Decompression, DecompressionLayer};
+use tracing::{debug, error, instrument};
+
+use super::auth::{AuthClientLayer, AuthClientService};
+
+#[derive(Clone, Debug)]
+pub(crate) struct HttpClient {
+    client: Decompression<AuthClientService<Client<HttpsConnector<HttpConnector>, HttpBody>>>,
+    url: Uri,
+    target: PayloadSource,
+}
+
+impl HttpClient {
+    pub(crate) fn new(url: Uri, secret: JwtSecret, target: PayloadSource) -> Self {
+        let connector = hyper_rustls::HttpsConnectorBuilder::new()
+            .with_native_roots()
+            .expect("no native root CA certificates found")
+            .https_or_http()
+            .enable_http1()
+            .enable_http2()
+            .build();
+
+        let client = Client::builder(TokioExecutor::new()).build(connector);
+
+        let client = ServiceBuilder::new()
+            .layer(DecompressionLayer::new())
+            .layer(AuthClientLayer::new(secret))
+            .service(client);
+
+        Self {
+            client,
+            url,
+            target,
+        }
+    }
+
+    /// Forwards an HTTP request to the `authrpc`, attaching the provided JWT authorization.
+    #[instrument(
+        skip(self, req),
+        fields(otel.kind = ?SpanKind::Client),
+        err(Debug)
+    )]
+    pub async fn forward(
+        &mut self,
+        mut req: http::Request<HttpBody>,
+        method: String,
+    ) -> Result<http::Response<HttpBody>, BoxError> {
+        debug!("forwarding {} to {}", method, self.target);
+        *req.uri_mut() = self.url.clone();
+
+        let res = self.client.ready().await?.call(req).await?;
+
+        let (parts, body) = res.into_parts();
+        let body_bytes = body.collect().await?.to_bytes().to_vec();
+
+        if let Some(code) = parse_response_code(&body_bytes)? {
+            error!(%code, "error in forwarded response");
+        }
+
+        Ok(http::Response::from_parts(
+            parts,
+            HttpBody::from(body_bytes),
+        ))
+    }
+}
+
+fn parse_response_code(body_bytes: &[u8]) -> eyre::Result<Option<i32>> {
+    #[derive(serde::Deserialize, Debug)]
+    struct RpcResponse {
+        error: Option<JsonRpcError>,
+    }
+
+    #[derive(serde::Deserialize, Debug)]
+    struct JsonRpcError {
+        code: i32,
+    }
+
+    let res = serde_json::from_slice::<RpcResponse>(body_bytes)?;
+
+    Ok(res.error.map(|e| e.code))
+}

src/client/mod.rs

@@ -0,0 +1,3 @@
+pub mod auth;
+pub mod http;
+pub mod rpc;

src/client.rs → src/client/rpc.rs

@@ -1,4 +1,4 @@
-use crate::auth_layer::{AuthClientLayer, AuthClientService};
+use crate::client::auth::{AuthClientLayer, AuthClientService};
 use crate::server::{EngineApiClient, PayloadSource};
 use alloy_primitives::B256;
 use alloy_rpc_types_engine::{
@@ -20,10 +20,10 @@ use tracing::{error, info, instrument};
 
 const INTERNAL_ERROR: i32 = 13;
 
-pub(crate) type ClientResult<T> = Result<T, ClientError>;
+pub(crate) type ClientResult<T> = Result<T, RpcClientError>;
 
 #[derive(Error, Debug)]
-pub(crate) enum ClientError {
+pub(crate) enum RpcClientError {
     #[error(transparent)]
     Jsonrpsee(#[from] jsonrpsee::core::client::Error),
     #[error("Invalid payload: {0}")]
@@ -53,10 +53,10 @@ impl<T, E: Code> Code for Result<T, E> {
 }
 
 /// TODO: Add more robust error code system
-impl Code for ClientError {
+impl Code for RpcClientError {
     fn code(&self) -> i32 {
         match self {
-            ClientError::Jsonrpsee(e) => e.code(),
+            RpcClientError::Jsonrpsee(e) => e.code(),
             // Status code 13 == internal error
             _ => INTERNAL_ERROR,
         }
@@ -72,10 +72,10 @@ impl Code for jsonrpsee::core::client::Error {
     }
 }
 
-impl From<ClientError> for ErrorObjectOwned {
-    fn from(err: ClientError) -> Self {
+impl From<RpcClientError> for ErrorObjectOwned {
+    fn from(err: RpcClientError) -> Self {
         match err {
-            ClientError::Jsonrpsee(jsonrpsee::core::ClientError::Call(error_object)) => {
+            RpcClientError::Jsonrpsee(jsonrpsee::core::ClientError::Call(error_object)) => {
                 error_object
             }
             // Status code 13 == internal error
@@ -89,7 +89,7 @@ impl From<ClientError> for ErrorObjectOwned {
 /// - **Engine API** calls are faciliated via the `auth_client` (requires JWT authentication).
 ///
 #[derive(Clone)]
-pub(crate) struct ExecutionClient {
+pub(crate) struct RpcClient {
     /// Handles requests to the authenticated Engine API (requires JWT authentication)
     auth_client: HttpClient<AuthClientService<HttpBackend>>,
     /// Uri of the RPC server for authenticated Engine API calls
@@ -98,14 +98,14 @@ pub(crate) struct ExecutionClient {
     payload_source: PayloadSource,
 }
 
-impl ExecutionClient {
+impl RpcClient {
     /// Initializes a new [ExecutionClient] with JWT auth for the Engine API and without auth for general execution layer APIs.
     pub fn new(
         auth_rpc: Uri,
         auth_rpc_jwt_secret: JwtSecret,
         timeout: u64,
         payload_source: PayloadSource,
-    ) -> Result<Self, ClientError> {
+    ) -> Result<Self, RpcClientError> {
         let auth_layer = AuthClientLayer::new(auth_rpc_jwt_secret);
         let auth_client = HttpClientBuilder::new()
             .set_http_middleware(tower::ServiceBuilder::new().layer(auth_layer))
@@ -148,7 +148,7 @@ impl ExecutionClient {
         }
 
         if res.is_invalid() {
-            return Err(ClientError::InvalidPayload(
+            return Err(RpcClientError::InvalidPayload(
                 res.payload_status.status.to_string(),
             ))
             .set_code();
@@ -208,7 +208,7 @@ impl ExecutionClient {
             .set_code()?;
 
         if res.is_invalid() {
-            return Err(ClientError::InvalidPayload(res.status.to_string()).set_code());
+            return Err(RpcClientError::InvalidPayload(res.status.to_string()).set_code());
         }
 
         Ok(res)
@@ -251,7 +251,7 @@ mod tests {
     use http::Uri;
     use jsonrpsee::core::client::ClientT;
 
-    use crate::auth_layer::AuthClientService;
+    use crate::client::auth::AuthClientService;
     use crate::server::PayloadSource;
     use alloy_rpc_types_engine::JwtSecret;
     use jsonrpsee::RpcModule;
@@ -290,7 +290,7 @@ mod tests {
         let secret = JwtSecret::from_hex(SECRET).unwrap();
 
         let auth_rpc = Uri::from_str(&format!("http://{}:{}", AUTH_ADDR, AUTH_PORT)).unwrap();
-        let client = ExecutionClient::new(auth_rpc, secret, 1000, PayloadSource::L2).unwrap();
+        let client = RpcClient::new(auth_rpc, secret, 1000, PayloadSource::L2).unwrap();
         let response = send_request(client.auth_client).await;
         assert!(response.is_ok());
         assert_eq!(response.unwrap(), "You are the dark lord");
@@ -300,7 +300,7 @@ mod tests {
     async fn invalid_jwt() {
         let secret = JwtSecret::random();
         let auth_rpc = Uri::from_str(&format!("http://{}:{}", AUTH_ADDR, AUTH_PORT)).unwrap();
-        let client = ExecutionClient::new(auth_rpc, secret, 1000, PayloadSource::L2).unwrap();
+        let client = RpcClient::new(auth_rpc, secret, 1000, PayloadSource::L2).unwrap();
         let response = send_request(client.auth_client).await;
         assert!(response.is_err());
         assert!(matches!(

src/health.rs

@@ -0,0 +1,67 @@
+use std::{
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+use futures::FutureExt as _;
+use jsonrpsee::{
+    core::BoxError,
+    http_client::{HttpBody, HttpRequest, HttpResponse},
+};
+use tower::{Layer, Service, util::Either};
+
+/// A [`Layer`] that filters out /healthz requests and responds with a 200 OK.
+#[derive(Clone, Debug)]
+pub(crate) struct HealthLayer;
+
+impl<S> Layer<S> for HealthLayer {
+    type Service = HealthService<S>;
+
+    fn layer(&self, inner: S) -> Self::Service {
+        HealthService { inner }
+    }
+}
+
+#[derive(Clone, Debug)]
+pub struct HealthService<S> {
+    inner: S,
+}
+
+impl<S> Service<HttpRequest<HttpBody>> for HealthService<S>
+where
+    S: Service<HttpRequest<HttpBody>, Response = HttpResponse> + Send + Sync + Clone + 'static,
+    S::Response: 'static,
+    S::Error: Into<BoxError> + 'static,
+    S::Future: Send + 'static,
+{
+    type Response = HttpResponse;
+    type Error = BoxError;
+    type Future = Either<
+        Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>,
+        S::Future,
+    >;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.inner.poll_ready(cx).map_err(Into::into)
+    }
+
+    fn call(&mut self, request: HttpRequest<HttpBody>) -> Self::Future {
+        if request.uri().path() == "/healthz" {
+            Either::A(Self::healthz().boxed())
+        } else {
+            Either::B(self.inner.call(request))
+        }
+    }
+}
+
+impl<S> HealthService<S>
+where
+    S: Service<HttpRequest<HttpBody>, Response = HttpResponse> + Send + Sync + Clone + 'static,
+    S::Response: 'static,
+    S::Error: Into<BoxError> + 'static,
+    S::Future: Send + 'static,
+{
+    async fn healthz() -> Result<HttpResponse, BoxError> {
+        Ok(HttpResponse::new(HttpBody::from("OK")))
+    }
+}

src/integration/mod.rs

@@ -1,4 +1,4 @@
-use crate::auth_layer::{AuthClientLayer, AuthClientService};
+use crate::client::auth::{AuthClientLayer, AuthClientService};
 use crate::debug_api::DebugClient;
 use crate::server::EngineApiClient;
 use crate::server::PayloadSource;