Merge pull request #5 from flashbots/proxy

Add proxy layer to forward calls to op-geth

.env.example

@@ -0,0 +1,3 @@
+JWT_TOKEN=688f5d737bad920bdfb2fc2f488d6b6209eebda1dae949a8de91398d932c517a
+L2_URL=http://localhost:8551
+BUILDER_URL=http://localhost:8552

.github/workflows/lint.yml

@@ -0,0 +1,31 @@
+name: Linting
+
+on:
+    push:
+      branches:
+        - main
+    pull_request:
+
+jobs:
+    build:
+      runs-on: ubuntu-latest
+
+      steps:
+        - name: Checkout code
+          uses: actions/checkout@v2
+
+        - name: Set up Rust
+          uses: actions-rs/toolchain@v1
+          with:
+            toolchain: stable
+            override: true
+            components: rustfmt
+
+        - name: Build
+          run: cargo build --verbose
+
+        - name: Lint
+          run: cargo clippy -- -D warnings
+
+        - name: Format code
+          run: cargo fmt -- --check

Cargo.toml

@@ -17,6 +17,11 @@ clap = { version = "4", features = ["derive", "env"] }
 jsonrpsee = {version = "0.24.4", features = ["server", "http-client", "macros"]}
 reqwest = "0.12.5"
 http = "1.1.0"
+dotenv = "0.15.0"
+tower = "0.4.13"
+http-body = "0.4.5"
+hyper = { version = "1.4.1", features = ["full"] }
+hyper-util = { version = "0.1", features = ["full"] }
 
 [dev-dependencies]
 anyhow = "1.0"

src/main.rs

@@ -2,17 +2,20 @@ use std::net::SocketAddr;
 use std::sync::Arc;
 
 use clap::{arg, Parser};
+use dotenv::dotenv;
 use error::Error;
 use http::HeaderMap;
 use http::{header::AUTHORIZATION, HeaderValue};
 use jsonrpsee::http_client::HttpClientBuilder;
 use jsonrpsee::server::Server;
 use jsonrpsee::RpcModule;
+use proxy::ProxyLayer;
 use server::{EthApiServer, EthEngineApi};
 use tracing::{info, Level};
 use tracing_subscriber::EnvFilter;
 
 mod error;
+mod proxy;
 mod server;
 
 #[derive(Parser, Debug)]
@@ -51,6 +54,8 @@ type Result<T> = core::result::Result<T, Error>;
 
 #[tokio::main]
 async fn main() -> Result<()> {
+    // Load .env file
+    dotenv().ok();
     let args: Args = Args::parse();
 
     // Initialize logging
@@ -69,7 +74,7 @@ async fn main() -> Result<()> {
     );
     let l2_client = HttpClientBuilder::new()
         .set_headers(headers.clone())
-        .build(args.l2_url)
+        .build(args.l2_url.clone())
         .unwrap();
 
     let builder_client = HttpClientBuilder::new()
@@ -78,14 +83,16 @@ async fn main() -> Result<()> {
         .unwrap();
 
     let eth_engine_api = EthEngineApi::new(Arc::new(l2_client), Arc::new(builder_client));
-    let mut module = RpcModule::new(());
+    let mut module: RpcModule<()> = RpcModule::new(());
     module
         .merge(eth_engine_api.into_rpc())
         .map_err(|e| Error::InitRPCServerError(e.to_string()))?;
 
     // server setup
     info!("Starting server on :{}", args.rpc_port);
+    let service_builder = tower::ServiceBuilder::new().layer(ProxyLayer::new(args.l2_url));
     let server = Server::builder()
+        .set_http_middleware(service_builder)
         .build(
             format!("{}:{}", args.rpc_host, args.rpc_port)
                 .parse::<SocketAddr>()
@@ -94,7 +101,7 @@ async fn main() -> Result<()> {
         .await
         .map_err(|e| Error::InitRPCServerError(e.to_string()))?;
     let handle = server.start(module);
-    tokio::spawn(handle.stopped());
+    handle.stopped().await;
 
     Ok(())
 }

src/proxy.rs

@@ -0,0 +1,70 @@
+use hyper::Response;
+use hyper_util::client::legacy::connect::HttpConnector;
+use hyper_util::client::legacy::Client;
+use hyper_util::rt::TokioExecutor;
+use jsonrpsee::core::BoxError;
+use jsonrpsee::http_client::{HttpBody, HttpRequest, HttpResponse};
+use std::task::{Context, Poll};
+use std::{future::Future, pin::Pin};
+use tower::{Layer, Service};
+use tracing::error;
+
+#[derive(Debug, Clone)]
+pub struct ProxyLayer {
+    target_url: String,
+}
+
+impl ProxyLayer {
+    pub fn new(target_url: String) -> Self {
+        ProxyLayer { target_url }
+    }
+}
+
+impl<S> Layer<S> for ProxyLayer {
+    type Service = ProxyService<S>;
+
+    fn layer(&self, inner: S) -> Self::Service {
+        ProxyService {
+            inner,
+            client: Client::builder(TokioExecutor::new()).build_http(),
+            target_url: self.target_url.clone(),
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct ProxyService<S> {
+    inner: S,
+    client: Client<HttpConnector, HttpBody>,
+    target_url: String,
+}
+
+impl<S> Service<HttpRequest<HttpBody>> for ProxyService<S>
+where
+    S: Service<HttpRequest<HttpBody>, Response = Response<HttpBody>>,
+    S::Response: 'static,
+    S::Error: Into<BoxError> + 'static,
+    S::Future: Send + 'static,
+{
+    type Response = HttpResponse<hyper::body::Incoming>;
+    type Error = BoxError;
+    type Future =
+        Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;
+
+    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, mut req: HttpRequest<HttpBody>) -> Self::Future {
+        let target_url = self.target_url.clone();
+        let client = self.client.clone();
+        let fut = async move {
+            *req.uri_mut() = target_url.parse().unwrap();
+            client.request(req).await.map_err(|e| {
+                error!("Error proxying request: {}", e);
+                Box::new(e) as Box<dyn std::error::Error + Send + Sync>
+            })
+        };
+        Box::pin(fut)
+    }
+}