Handle Inconsistent Response in Miner Api

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};
 
@@ -157,11 +158,12 @@ 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,
-            self.execution_mode,
-            probes,
+            execution_mode.clone(),
+            probes.clone(),
             self.health_check_interval,
             self.max_unsafe_interval,
         );
@@ -182,6 +184,8 @@ impl Args {
                     l2_auth_jwt,
                     builder_args.builder_url,
                     builder_auth_jwt,
+                    probes,
+                    execution_mode,
                 ));
 
         let server = Server::builder()

src/proxy.rs

@@ -1,20 +1,28 @@
 use crate::client::http::HttpClient;
 use crate::server::PayloadSource;
+use crate::{ExecutionMode, 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 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",
@@ -27,6 +35,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 {
@@ -35,12 +45,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,
         }
     }
 }
@@ -65,6 +79,8 @@ impl<S> Layer<S> for ProxyLayer {
             inner,
             l2_client,
             builder_client,
+            probes: self.probes.clone(),
+            execution_mode: self.execution_mode.clone(),
         }
     }
 }
@@ -74,6 +90,8 @@ pub struct ProxyService<S> {
     inner: S,
     l2_client: HttpClient,
     builder_client: HttpClient,
+    probes: Arc<Probes>,
+    execution_mode: Arc<Mutex<ExecutionMode>>,
 }
 
 // Consider using `RpcServiceT` when https://github.com/paritytech/jsonrpsee/pull/1521 is merged
@@ -131,8 +149,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::Disabled;
+                        // Drop before aquiring health lock
+                        drop(execution_mode);
+                        warn!(target: "proxy::call", message = "setting execution mode to Disabled");
+                        service.probes.set_health(Health::PartialContent);
+                    }
+                }
+                l2_res
             } else {
                 // If the request should not be forwarded, send directly to the
                 // default execution client
@@ -172,7 +214,7 @@ mod tests {
     use std::{
         net::{IpAddr, SocketAddr},
         str::FromStr,
-        sync::{Arc, Mutex},
+        sync::Arc,
     };
     use tokio::net::TcpListener;
     use tokio::task::JoinHandle;
@@ -198,12 +240,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?;
@@ -313,7 +358,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();
 
@@ -391,7 +436,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();
@@ -408,8 +454,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();
@@ -446,7 +493,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();
@@ -457,9 +504,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()
@@ -509,7 +563,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);
@@ -518,7 +572,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);
@@ -545,15 +599,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);
@@ -582,7 +636,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);
@@ -591,7 +645,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);
@@ -618,15 +672,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);
@@ -652,15 +706,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);
@@ -687,15 +741,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);
@@ -721,12 +775,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

@@ -134,7 +134,7 @@ impl RollupBoostServer {
     pub fn new(
         l2_client: RpcClient,
         builder_client: RpcClient,
-        initial_execution_mode: ExecutionMode,
+        initial_execution_mode: Arc<Mutex<ExecutionMode>>,
         probes: Arc<Probes>,
         health_check_interval: u64,
         max_unsafe_interval: u64,
@@ -151,7 +151,7 @@ impl RollupBoostServer {
             l2_client: Arc::new(l2_client),
             builder_client: Arc::new(builder_client),
             payload_trace_context: Arc::new(PayloadTraceContext::new()),
-            execution_mode: Arc::new(Mutex::new(initial_execution_mode)),
+            execution_mode: initial_execution_mode,
             probes,
             health_handle,
         }
@@ -808,12 +808,13 @@ 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,
-                ExecutionMode::Enabled,
-                probes,
+                execution_mode.clone(),
+                probes.clone(),
                 60,
                 5,
             );
@@ -828,6 +829,8 @@ mod tests {
                         jwt_secret,
                         builder_auth_rpc,
                         jwt_secret,
+                        probes,
+                        execution_mode.clone(),
                     ));
 
             let server = Server::builder()