Handle Inconsistent Response in Miner Api

Cargo.toml

@@ -18,7 +18,7 @@ jsonrpsee = { version = "0.24", features = ["server", "http-client", "macros"] }
 moka = { version = "0.12.10", features = ["future"] }
 http = "1.1.0"
 dotenvy = "0.15.7"
-tower = "0.4.13"
+tower = { version = "0.4.13", features = ["timeout"] }
 tower-http = { version = "0.5.2", features = [
   "decompression-full",
   "sensitive-headers",

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};
 
@@ -159,12 +160,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,
-            self.execution_mode,
+            execution_mode.clone(),
             self.block_selection_policy,
-            probes,
+            probes.clone(),
             self.health_check_interval,
             self.max_unsafe_interval,
         );
@@ -185,6 +187,8 @@ impl Args {
                     l2_auth_jwt,
                     builder_args.builder_url,
                     builder_auth_jwt,
+                    probes,
+                    execution_mode,
                 ));
 
         let server = Server::builder()

src/client/http.rs

@@ -1,22 +1,30 @@
+use std::time::Duration;
+
 use crate::client::auth::AuthLayer;
 use crate::payload::PayloadSource;
+use alloy_primitives::bytes::Bytes;
 use alloy_rpc_types_engine::JwtSecret;
 use http::Uri;
-use http_body_util::BodyExt;
+use http_body_util::{BodyExt, Full};
+use hyper::body::Body;
 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 jsonrpsee::server::HttpBody;
 use opentelemetry::trace::SpanKind;
-use tower::{Service as _, ServiceBuilder, ServiceExt};
+use tower::{
+    Service as _, ServiceBuilder, ServiceExt,
+    timeout::{Timeout, TimeoutLayer},
+};
 use tower_http::decompression::{Decompression, DecompressionLayer};
 use tracing::{debug, error, instrument};
 
 use super::auth::Auth;
 
-pub type HttpClientService = Decompression<Auth<Client<HttpsConnector<HttpConnector>, HttpBody>>>;
+pub type HttpClientService =
+    Timeout<Decompression<Auth<Client<HttpsConnector<HttpConnector>, HttpBody>>>>;
 
 #[derive(Clone, Debug)]
 pub struct HttpClient {
@@ -38,6 +46,7 @@ impl HttpClient {
         let client = Client::builder(TokioExecutor::new()).build(connector);
 
         let client = ServiceBuilder::new()
+            .layer(TimeoutLayer::new(Duration::from_secs(1)))
             .layer(DecompressionLayer::new())
             .layer(AuthLayer::new(secret))
             .service(client);
@@ -52,36 +61,41 @@ impl HttpClient {
     /// Forwards an HTTP request to the `authrpc`, attaching the provided JWT authorization.
     #[instrument(
         skip(self, req),
-        fields(otel.kind = ?SpanKind::Client,
-        url = %self.url,
-        method,
-        code,
+        fields(
+            otel.kind = ?SpanKind::Client,
+            url = %self.url,
+            method,
+            code,
         ),
         err(Debug)
     )]
-    pub async fn forward(
+    pub async fn forward<B>(
         &mut self,
-        mut req: http::Request<HttpBody>,
+        mut req: http::Request<B>,
         method: String,
-    ) -> Result<http::Response<HttpBody>, BoxError> {
+    ) -> Result<http::Response<Full<Bytes>>, BoxError>
+    where
+        B: Body<Data = Bytes, Error: Into<Box<dyn std::error::Error + Send + Sync>>>
+            + Send
+            + 'static,
+    {
         debug!("forwarding {} to {}", method, self.target);
         tracing::Span::current().record("method", method);
         *req.uri_mut() = self.url.clone();
 
+        let req = req.map(HttpBody::new);
+
         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();
+        let body_bytes = body.collect().await?.to_bytes();
 
         if let Some(code) = parse_response_code(&body_bytes)? {
             error!(%code, "error in forwarded response");
             tracing::Span::current().record("code", code);
         }
 
-        Ok(http::Response::from_parts(
-            parts,
-            HttpBody::from(body_bytes),
-        ))
+        Ok(http::Response::from_parts(parts, Full::from(body_bytes)))
     }
 }
 

src/consistent_request.rs

@@ -0,0 +1,195 @@
+//! consistent_request.rs
+use std::{
+    sync::{
+        Arc,
+        atomic::{AtomicBool, Ordering},
+    },
+    time::Duration,
+};
+
+use eyre::{Result as EyreResult, bail};
+use jsonrpsee::{core::BoxError, http_client::HttpBody};
+use parking_lot::Mutex;
+use tokio::{sync::watch, task::JoinHandle};
+use tracing::{error, info, warn};
+
+use crate::{
+    BufferedRequest, BufferedResponse, ExecutionMode, Health, HttpClient, Probes, Response,
+};
+
+/// A request manager that ensures requests are sent consistently
+/// across both the builder and l2 client.
+#[derive(Clone)]
+pub struct ConsistentRequest {
+    method: String,
+    l2_client: HttpClient,
+    builder_client: HttpClient,
+    has_disabled_execution_mode: Arc<AtomicBool>,
+    req_tx: watch::Sender<Option<BufferedRequest>>,
+    res_rx: watch::Receiver<Option<Result<BufferedResponse, BoxError>>>,
+    probes: Arc<Probes>,
+    execution_mode: Arc<Mutex<ExecutionMode>>,
+}
+
+impl ConsistentRequest {
+    pub fn new(
+        method: String,
+        l2_client: HttpClient,
+        builder_client: HttpClient,
+        probes: Arc<Probes>,
+        execution_mode: Arc<Mutex<ExecutionMode>>,
+    ) -> Self {
+        let (req_tx, mut req_rx) = watch::channel(None);
+        let (res_tx, mut res_rx) = watch::channel(None);
+        req_rx.mark_unchanged();
+        res_rx.mark_unchanged();
+
+        let has_disabled_execution_mode = Arc::new(AtomicBool::new(false));
+
+        let manager = Self {
+            method,
+            l2_client,
+            builder_client,
+            has_disabled_execution_mode,
+            req_tx,
+            res_rx,
+            probes,
+            execution_mode,
+        };
+
+        let clone = manager.clone();
+        tokio::spawn(async move {
+            let mut attempt: Option<JoinHandle<_>> = None;
+            loop {
+                req_rx
+                    .changed()
+                    .await
+                    .expect("channel should always be open");
+
+                if let Some(attempt) = attempt {
+                    if !attempt.is_finished() {
+                        error!(
+                            target: "proxy::call",
+                            method = clone.method,
+                            "request cancelled"
+                        );
+                        attempt.abort();
+                    }
+                }
+
+                let req = req_rx
+                    .borrow_and_update()
+                    .as_ref()
+                    .expect("value should always be Some")
+                    .clone();
+
+                let mut clone = clone.clone();
+                let res_tx_clone = res_tx.clone();
+                attempt = Some(tokio::spawn(async move {
+                    clone.send_with_retry_cancel_safe(req, res_tx_clone).await
+                }));
+            }
+        });
+
+        manager
+    }
+
+    /// This function may be cancelled at any time from an incoming request.
+    async fn send_with_retry_cancel_safe(
+        &mut self,
+        req: BufferedRequest,
+        res_tx: watch::Sender<Option<Result<BufferedResponse, BoxError>>>,
+    ) -> EyreResult<()> {
+        // We send the l2 request first, because we need to avoid the situation where the
+        // l2 fails and the builder succeeds. If this were to happen, it would be too dangerous to
+        // return the l2 error response back to the caller, since the builder would now have an
+        // invalid state. We can return early if the l2 request fails. Note we're specifically
+        // spawning new tasks here to avoid any issues with cancellation. We should ensure we're
+        // in a valid state at all await points.
+        let mut manager = self.clone();
+        let res_tx_clone = res_tx.clone();
+        let req_clone = req.clone();
+        tokio::spawn(async move { manager.send_to_l2(req_clone, res_tx_clone).await }).await??;
+
+        loop {
+            let mut manager_clone = self.clone();
+            let req_clone = req.clone();
+
+            match tokio::spawn(async move { manager_clone.send_to_builder(req_clone).await })
+                .await?
+            {
+                Ok(_) => return Ok(()),
+                Err(_) => tokio::time::sleep(Duration::from_millis(200)).await,
+            }
+        }
+    }
+
+    async fn send_to_l2(
+        &mut self,
+        req: BufferedRequest,
+        res_tx: watch::Sender<Option<Result<BufferedResponse, BoxError>>>,
+    ) -> EyreResult<()> {
+        let l2_res = self.l2_client.forward(req, self.method.clone()).await;
+        match l2_res {
+            Ok(_) => {
+                res_tx.send(Some(l2_res))?;
+                Ok(())
+            }
+            Err(_) => {
+                res_tx.send(Some(l2_res))?;
+                Err(eyre::eyre!("failed to send request to L2 client"))
+            }
+        }
+    }
+
+    async fn send_to_builder(&mut self, req: BufferedRequest) -> EyreResult<()> {
+        match self.builder_client.forward(req, self.method.clone()).await {
+            Ok(_) => {
+                if self.has_disabled_execution_mode.load(Ordering::SeqCst) {
+                    let mut mode = self.execution_mode.lock();
+                    *mode = ExecutionMode::Enabled;
+                    drop(mode);
+                    self.has_disabled_execution_mode
+                        .store(false, Ordering::SeqCst);
+                    info!(target: "proxy::call", message = "setting execution mode to Enabled");
+                }
+                Ok(())
+            }
+            Err(e) => {
+                // l2 request succeeded, but builder request failed
+                // This state can only be recovered from if either the builder is restarted
+                // or if a retry eventually goes through.
+                error!(target: "proxy::call", method = self.method, "inconsistent responses from builder and L2");
+                let mut mode = self.execution_mode.lock();
+                if *mode == ExecutionMode::Enabled {
+                    *mode = ExecutionMode::Disabled;
+                    // Drop before aquiring health lock
+                    drop(mode);
+                    self.has_disabled_execution_mode
+                        .store(true, Ordering::SeqCst);
+                    warn!(target: "proxy::call", "setting execution mode to Disabled");
+                    // This health status will likely be later set back to healthy
+                    // but this should be enough to trigger a new leader election.
+                    self.probes.set_health(Health::PartialContent);
+                }
+
+                bail!("failed to send request to builder: {e}");
+            }
+        }
+    }
+
+    // Send a request, ensuring consistent responses from both the builder and l2 client.
+    pub async fn send(&mut self, req: BufferedRequest) -> Result<Response, BoxError> {
+        self.req_tx.send(Some(req))?;
+        self.res_rx.changed().await?;
+        match self
+            .res_rx
+            .borrow_and_update()
+            .as_ref()
+            .expect("value should always be Some")
+        {
+            Ok(v) => Ok(v.clone().map(HttpBody::new)),
+            Err(e) => Err(format!("error sending consistent request: {e}").into()),
+        }
+    }
+}

src/lib.rs

@@ -1,3 +1,5 @@
+#![allow(clippy::complexity)]
+
 mod client;
 pub use client::{auth::*, http::*, rpc::*};
 
@@ -30,3 +32,5 @@ pub use payload::*;
 
 mod selection;
 pub use selection::*;
+
+mod consistent_request;