refactor

Signed-off-by: Naohiro Yoshida <naohiro.yoshida@datachain.jp>

Author: Naohiro Yoshida

server/src/host/single/handler.rs

@@ -3,7 +3,7 @@
 use crate::host::single::config::Config;
 use crate::host::single::local_kv::LocalKeyValueStore;
 use crate::host::single::trace::{encode_to_bytes, TracingKeyValueStore};
-use crate::transport::Http;
+use crate::transport::Transport;
 use alloy_primitives::B256;
 use alloy_provider::RootProvider;
 use alloy_rpc_client::RpcClient;
@@ -21,7 +21,6 @@ use kona_proof::boot::L2_ROLLUP_CONFIG_KEY;
 use kona_proof::HintType;
 use kona_providers_alloy::{OnlineBeaconClient, OnlineBlobProvider};
 use op_alloy_network::{Network, Optimism};
-use reqwest::Client;
 use std::sync::Arc;
 use tokio::sync::RwLock;
 
@@ -67,7 +66,7 @@ impl DerivationRequest {
 
     fn http_provider<N: Network>(url: &str) -> RootProvider<N> {
         let url = url.parse().unwrap();
-        let http = Http::<Client>::new(url);
+        let http = Transport::new(url);
         RootProvider::new(RpcClient::new(http, true))
     }
 

server/src/transport/hash.rs

@@ -0,0 +1,52 @@
+use alloy_json_rpc::{RequestPacket, ResponsePacket};
+use alloy_primitives::{keccak256, B256};
+use alloy_transport::{TransportError, TransportFut};
+use std::task;
+use tower::Service;
+
+use crate::transport::Transport;
+
+impl Service<RequestPacket> for Transport {
+    type Response = ResponsePacket;
+    type Error = TransportError;
+    type Future = TransportFut<'static>;
+
+    fn poll_ready(&mut self, _cx: &mut task::Context<'_>) -> task::Poll<Result<(), Self::Error>> {
+        task::Poll::Ready(Ok(()))
+    }
+
+    fn call(&mut self, req: RequestPacket) -> Self::Future {
+        let hash = match from_request_to_hash(&req) {
+            Ok(hash) => hash,
+            Err(err) => {
+                return Box::pin(async move {
+                    Err(TransportError::deser_err(
+                        err,
+                        "invalid request for hashing",
+                    ))
+                });
+            }
+        };
+        let mut headers = req.headers();
+        headers.insert("X-Request-Hash", hash.to_string().parse().unwrap());
+
+        let this = self.clone();
+        Box::pin(this.post(req, headers))
+    }
+}
+
+fn from_request_to_hash(req: &RequestPacket) -> Result<B256, serde_json::Error> {
+    let body = match &req {
+        RequestPacket::Single(r) => serde_json::to_vec(r)?,
+        RequestPacket::Batch(_) => {
+            let mut value = vec![];
+            if let Some(batch) = req.as_batch() {
+                for r in batch {
+                    value.extend_from_slice(&r.params_hash().0);
+                }
+            }
+            value
+        }
+    };
+    Ok(keccak256(&body))
+}

server/src/transport/mod.rs

@@ -1,9 +1,51 @@
-pub mod reqwest_transport;
+pub mod hash;
 
+use alloy_json_rpc::{RequestPacket, ResponsePacket};
+use alloy_transport::{TransportError, TransportErrorKind, TransportResult};
+use reqwest::header::HeaderMap;
+use reqwest::Client;
+use tracing::{debug, trace};
 use url::Url;
 
 #[derive(Clone, Debug)]
-pub struct Http<T> {
-    client: T,
+pub struct Transport {
+    client: Client,
     url: Url,
 }
+
+impl Transport {
+    pub fn new(url: Url) -> Self {
+        Self {
+            client: Default::default(),
+            url,
+        }
+    }
+
+    async fn post(self, req: RequestPacket, headers: HeaderMap) -> TransportResult<ResponsePacket> {
+        let resp = self
+            .client
+            .post(self.url)
+            .json(&req)
+            .headers(headers)
+            .send()
+            .await
+            .map_err(TransportErrorKind::custom)?;
+        let status = resp.status();
+
+        debug!(%status, "received response from server");
+
+        let body = resp.bytes().await.map_err(TransportErrorKind::custom)?;
+
+        trace!(body = %String::from_utf8_lossy(&body), "response body");
+
+        if !status.is_success() {
+            return Err(TransportErrorKind::http_error(
+                status.as_u16(),
+                String::from_utf8_lossy(&body).into_owned(),
+            ));
+        }
+
+        serde_json::from_slice(&body)
+            .map_err(|err| TransportError::deser_err(err, String::from_utf8_lossy(&body)))
+    }
+}