Handle Inconsistent Response in Miner Api

Author: 0xForerunner

src/cli.rs

@@ -1,9 +1,10 @@
-use std::{net::SocketAddr, path::PathBuf};
+use std::{net::SocketAddr, path::PathBuf, sync::Arc};
 
 use alloy_rpc_types_engine::JwtSecret;
 use clap::{Parser, Subcommand};
 use eyre::bail;
 use jsonrpsee::{RpcModule, server::Server};
+use parking_lot::Mutex;
 use tokio::signal::unix::{SignalKind, signal as unix_signal};
 use tracing::{Level, info};
 
@@ -166,12 +167,13 @@ impl Args {
 
         let (probe_layer, probes) = ProbeLayer::new();
 
+        let execution_mode = Arc::new(Mutex::new(self.execution_mode));
         let rollup_boost = RollupBoostServer::new(
             l2_client,
             builder_client,
             boost_sync_enabled,
-            self.execution_mode,
-            probes,
+            execution_mode.clone(),
+            probes.clone(),
             self.health_check_interval,
             self.max_unsafe_interval,
         );
@@ -192,6 +194,8 @@ impl Args {
                     l2_auth_jwt,
                     builder_args.builder_url,
                     builder_auth_jwt,
+                    probes,
+                    execution_mode,
                 ));
 
         let server = Server::builder()

src/proxy.rs

@@ -1,22 +1,30 @@
 use crate::client::http::HttpClient;
-use crate::server::PayloadSource;
+use crate::server::{ExecutionMode, PayloadSource};
+use crate::{Health, Probes};
 use alloy_rpc_types_engine::JwtSecret;
 use http::Uri;
 use jsonrpsee::core::{BoxError, http_helpers};
 use jsonrpsee::http_client::{HttpBody, HttpRequest, HttpResponse};
+use parking_lot::Mutex;
+use std::sync::Arc;
 use std::task::{Context, Poll};
 use std::{future::Future, pin::Pin};
 use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
 use tokio::sync::oneshot;
 use tower::{Layer, Service};
-use tracing::info;
+use tracing::{error, info, warn};
 
 const ENGINE_METHOD: &str = "engine_";
 
 /// Requests that should be forwarded to both the builder and default execution client
-const FORWARD_REQUESTS: [&str; 6] = [
+const FORWARD_REQUESTS: [&str; 2] = [
     "eth_sendRawTransaction",
     "eth_sendRawTransactionConditional",
+];
+
+/// Handle these similar to FORWARD_REQUESTS, but enforce consistant responses
+/// between the builder and default execution client.
+const MINER_REQUESTS: [&str; 4] = [
     "miner_setExtra",
     "miner_setGasPrice",
     "miner_setGasLimit",
@@ -29,6 +37,8 @@ pub struct ProxyLayer {
     l2_auth_secret: JwtSecret,
     builder_auth_rpc: Uri,
     builder_auth_secret: JwtSecret,
+    probes: Arc<Probes>,
+    execution_mode: Arc<Mutex<ExecutionMode>>,
 }
 
 impl ProxyLayer {
@@ -37,12 +47,16 @@ impl ProxyLayer {
         l2_auth_secret: JwtSecret,
         builder_auth_rpc: Uri,
         builder_auth_secret: JwtSecret,
+        probes: Arc<Probes>,
+        execution_mode: Arc<Mutex<ExecutionMode>>,
     ) -> Self {
         ProxyLayer {
             l2_auth_rpc,
             l2_auth_secret,
             builder_auth_rpc,
             builder_auth_secret,
+            probes,
+            execution_mode,
         }
     }
 }
@@ -75,6 +89,8 @@ where
             l2_client,
             builder_client,
             engine_tx: tx,
+            probes: self.probes.clone(),
+            execution_mode: self.execution_mode.clone(),
         };
 
         service.process_engine_queue(rx);
@@ -98,6 +114,8 @@ pub struct ProxyService<S> {
     l2_client: HttpClient,
     builder_client: HttpClient,
     engine_tx: EngineResponseTx,
+    probes: Arc<Probes>,
+    execution_mode: Arc<Mutex<ExecutionMode>>,
 }
 
 impl<S> ProxyService<S>
@@ -179,8 +197,32 @@ where
                 });
 
                 let l2_req = HttpRequest::from_parts(parts, HttpBody::from(body_bytes));
-                info!(target: "proxy::call", message = "forward request to default execution client", ?method);
                 service.l2_client.forward(l2_req, method).await
+            } else if MINER_REQUESTS.contains(&method.as_str()) {
+                // miner api, send to both the
+                // default execution client and the builder
+                let builder_req =
+                    HttpRequest::from_parts(parts.clone(), HttpBody::from(body_bytes.clone()));
+                let builder_method = method.clone();
+                let mut builder_client = service.builder_client.clone();
+
+                let l2_req = HttpRequest::from_parts(parts, HttpBody::from(body_bytes));
+                let (builder_res, l2_res) = tokio::join!(
+                    builder_client.forward(builder_req, builder_method),
+                    service.l2_client.forward(l2_req, method)
+                );
+                if builder_res.is_ok() != l2_res.is_ok() {
+                    error!(target: "proxy::call", message = "inconsistent miner api responses from builder and L2");
+                    let mut execution_mode = service.execution_mode.lock();
+                    if *execution_mode == ExecutionMode::Enabled {
+                        *execution_mode = ExecutionMode::Fallback;
+                        // Drop before aquiring health lock
+                        drop(execution_mode);
+                        warn!(target: "proxy::call", message = "setting execution mode to Fallback");
+                        service.probes.set_health(Health::PartialContent);
+                    }
+                }
+                l2_res
             } else {
                 // If the request should not be forwarded, send directly to the
                 // default execution client
@@ -220,7 +262,7 @@ mod tests {
     use std::{
         net::{IpAddr, SocketAddr},
         str::FromStr,
-        sync::{Arc, Mutex},
+        sync::Arc,
     };
     use tokio::net::TcpListener;
     use tokio::task::JoinHandle;
@@ -246,12 +288,15 @@ mod tests {
         async fn new() -> eyre::Result<Self> {
             let builder = MockHttpServer::serve().await?;
             let l2 = MockHttpServer::serve().await?;
+            let execution_mode = Arc::new(Mutex::new(ExecutionMode::Enabled));
+            let probes = Arc::new(Probes::default());
             let middleware = tower::ServiceBuilder::new().layer(ProxyLayer::new(
                 format!("http://{}:{}", l2.addr.ip(), l2.addr.port()).parse::<Uri>()?,
                 JwtSecret::random(),
                 format!("http://{}:{}", builder.addr.ip(), builder.addr.port()).parse::<Uri>()?,
                 JwtSecret::random(),
-                // None,
+                probes.clone(),
+                execution_mode.clone(),
             ));
 
             let temp_listener = TcpListener::bind("0.0.0.0:0").await?;
@@ -361,7 +406,7 @@ mod tests {
                 }
             };
 
-            requests.lock().unwrap().push(request_body.clone());
+            requests.lock().push(request_body.clone());
 
             let method = request_body["method"].as_str().unwrap_or_default();
 
@@ -439,7 +484,8 @@ mod tests {
     }
 
     async fn health_check() {
-        let proxy_server = spawn_proxy_server().await;
+        let execution_mode = Arc::new(Mutex::new(ExecutionMode::Enabled));
+        let proxy_server = spawn_proxy_server(execution_mode).await;
         // Create a new HTTP client
         let client: Client<HttpConnector, HttpBody> =
             Client::builder(TokioExecutor::new()).build_http();
@@ -456,8 +502,9 @@ mod tests {
     }
 
     async fn send_request(method: &str) -> Result<String, ClientError> {
+        let execution_mode = Arc::new(Mutex::new(ExecutionMode::Enabled));
         let server = spawn_server().await;
-        let proxy_server = spawn_proxy_server().await;
+        let proxy_server = spawn_proxy_server(execution_mode).await;
         let proxy_client = HttpClient::builder()
             .build(format!("http://{ADDR}:{PORT}"))
             .unwrap();
@@ -494,7 +541,7 @@ mod tests {
     }
 
     /// Spawn a new RPC server with a proxy layer.
-    async fn spawn_proxy_server() -> ServerHandle {
+    async fn spawn_proxy_server(execution_mode: Arc<Mutex<ExecutionMode>>) -> ServerHandle {
         let addr = format!("{ADDR}:{PORT}");
 
         let jwt = JwtSecret::random();
@@ -505,9 +552,16 @@ mod tests {
         .parse::<Uri>()
         .unwrap();
 
-        let (probe_layer, _probes) = ProbeLayer::new();
+        let (probe_layer, probes) = ProbeLayer::new();
 
-        let proxy_layer = ProxyLayer::new(l2_auth_uri.clone(), jwt, l2_auth_uri, jwt);
+        let proxy_layer = ProxyLayer::new(
+            l2_auth_uri.clone(),
+            jwt,
+            l2_auth_uri,
+            jwt,
+            probes,
+            execution_mode,
+        );
 
         // Create a layered server
         let server = ServerBuilder::default()
@@ -557,7 +611,7 @@ mod tests {
 
         // Assert the builder received the correct payload
         let builder = &test_harness.builder;
-        let builder_requests = builder.requests.lock().unwrap();
+        let builder_requests = builder.requests.lock();
         let builder_req = builder_requests.first().unwrap();
         assert_eq!(builder_requests.len(), 1);
         assert_eq!(builder_req["method"], expected_method);
@@ -566,7 +620,7 @@ mod tests {
 
         // Assert the l2 received the correct payload
         let l2 = &test_harness.l2;
-        let l2_requests = l2.requests.lock().unwrap();
+        let l2_requests = l2.requests.lock();
         let l2_req = l2_requests.first().unwrap();
         assert_eq!(l2_requests.len(), 1);
         assert_eq!(l2_req["method"], expected_method);
@@ -593,15 +647,15 @@ mod tests {
 
         // Assert the builder received the correct payload
         let builder = &test_harness.builder;
-        let builder_requests = builder.requests.lock().unwrap();
+        let builder_requests = builder.requests.lock();
         let builder_req = builder_requests.first().unwrap();
         assert_eq!(builder_requests.len(), 1);
         assert_eq!(builder_req["method"], expected_method);
         assert_eq!(builder_req["params"][0], expected_tx);
 
         // Assert the l2 received the correct payload
         let l2 = &test_harness.l2;
-        let l2_requests = l2.requests.lock().unwrap();
+        let l2_requests = l2.requests.lock();
         let l2_req = l2_requests.first().unwrap();
         assert_eq!(l2_requests.len(), 1);
         assert_eq!(l2_req["method"], expected_method);
@@ -630,7 +684,7 @@ mod tests {
         let expected_conditionals = json!(transact_conditionals);
         // Assert the builder received the correct payload
         let builder = &test_harness.builder;
-        let builder_requests = builder.requests.lock().unwrap();
+        let builder_requests = builder.requests.lock();
         let builder_req = builder_requests.first().unwrap();
         assert_eq!(builder_requests.len(), 1);
         assert_eq!(builder_req["method"], expected_method);
@@ -639,7 +693,7 @@ mod tests {
 
         // Assert the l2 received the correct payload
         let l2 = &test_harness.l2;
-        let l2_requests = l2.requests.lock().unwrap();
+        let l2_requests = l2.requests.lock();
         let l2_req = l2_requests.first().unwrap();
         assert_eq!(l2_requests.len(), 1);
         assert_eq!(l2_req["method"], expected_method);
@@ -666,15 +720,15 @@ mod tests {
 
         // Assert the builder received the correct payload
         let builder = &test_harness.builder;
-        let builder_requests = builder.requests.lock().unwrap();
+        let builder_requests = builder.requests.lock();
         let builder_req = builder_requests.first().unwrap();
         assert_eq!(builder_requests.len(), 1);
         assert_eq!(builder_req["method"], expected_method);
         assert_eq!(builder_req["params"][0], expected_extra);
 
         // Assert the l2 received the correct payload
         let l2 = &test_harness.l2;
-        let l2_requests = l2.requests.lock().unwrap();
+        let l2_requests = l2.requests.lock();
         let l2_req = l2_requests.first().unwrap();
         assert_eq!(l2_requests.len(), 1);
         assert_eq!(l2_req["method"], expected_method);
@@ -700,15 +754,15 @@ mod tests {
 
         // Assert the builder received the correct payload
         let builder = &test_harness.builder;
-        let builder_requests = builder.requests.lock().unwrap();
+        let builder_requests = builder.requests.lock();
         let builder_req = builder_requests.first().unwrap();
         assert_eq!(builder_requests.len(), 1);
         assert_eq!(builder_req["method"], expected_method);
         assert_eq!(builder_req["params"][0], expected_price);
 
         // Assert the l2 received the correct payload
         let l2 = &test_harness.l2;
-        let l2_requests = l2.requests.lock().unwrap();
+        let l2_requests = l2.requests.lock();
         let l2_req = l2_requests.first().unwrap();
         assert_eq!(l2_requests.len(), 1);
         assert_eq!(l2_req["method"], expected_method);
@@ -735,15 +789,15 @@ mod tests {
 
         // Assert the builder received the correct payload
         let builder = &test_harness.builder;
-        let builder_requests = builder.requests.lock().unwrap();
+        let builder_requests = builder.requests.lock();
         let builder_req = builder_requests.first().unwrap();
         assert_eq!(builder_requests.len(), 1);
         assert_eq!(builder_req["method"], expected_method);
         assert_eq!(builder_req["params"][0], expected_price);
 
         // Assert the l2 received the correct payload
         let l2 = &test_harness.l2;
-        let l2_requests = l2.requests.lock().unwrap();
+        let l2_requests = l2.requests.lock();
         let l2_req = l2_requests.first().unwrap();
         assert_eq!(l2_requests.len(), 1);
         assert_eq!(l2_req["method"], expected_method);
@@ -769,12 +823,12 @@ mod tests {
 
         // Assert the builder has not received the payload
         let builder = &test_harness.builder;
-        let builder_requests = builder.requests.lock().unwrap();
+        let builder_requests = builder.requests.lock();
         assert_eq!(builder_requests.len(), 0);
 
         // Assert the l2 auth received the correct payload
         let l2 = &test_harness.l2;
-        let l2_requests = l2.requests.lock().unwrap();
+        let l2_requests = l2.requests.lock();
         let l2_req = l2_requests.first().unwrap();
         assert_eq!(l2_requests.len(), 1);
         assert_eq!(l2_req["method"], expected_method);

src/server.rs

@@ -174,7 +174,7 @@ impl RollupBoostServer {
         l2_client: RpcClient,
         builder_client: RpcClient,
         boost_sync: bool,
-        initial_execution_mode: ExecutionMode,
+        initial_execution_mode: Arc<Mutex<ExecutionMode>>,
         probes: Arc<Probes>,
         health_check_interval: u64,
         max_unsafe_interval: u64,
@@ -192,7 +192,7 @@ impl RollupBoostServer {
             builder_client: Arc::new(builder_client),
             boost_sync,
             payload_trace_context: Arc::new(PayloadTraceContext::new()),
-            execution_mode: Arc::new(Mutex::new(initial_execution_mode)),
+            execution_mode: initial_execution_mode,
             probes,
             health_handle,
         }
@@ -824,13 +824,14 @@ mod tests {
             .unwrap();
 
             let (probe_layer, probes) = ProbeLayer::new();
+            let execution_mode = Arc::new(Mutex::new(ExecutionMode::Enabled));
 
             let rollup_boost = RollupBoostServer::new(
                 l2_client,
                 builder_client,
                 boost_sync,
-                ExecutionMode::Enabled,
-                probes,
+                execution_mode.clone(),
+                probes.clone(),
                 60,
                 5,
             );
@@ -845,6 +846,8 @@ mod tests {
                         jwt_secret,
                         builder_auth_rpc,
                         jwt_secret,
+                        probes,
+                        execution_mode.clone(),
                     ));
 
             let server = Server::builder()