Improve websockets reconnects

Cargo.toml

@@ -5,6 +5,7 @@ members = [
   "chains/bitcoin/config",
   "chains/bitcoin/server",
   "chains/ethereum/config",
+  "chains/ethereum/rpc-client",
   "chains/ethereum/server",
   "chains/ethereum/tx",
   "chains/polkadot/config",

chains/ethereum/rpc-client/Cargo.toml

@@ -0,0 +1,22 @@
+[package]
+name = "rosetta-ethereum-rpc-client"
+version = "0.1.0"
+edition = "2021"
+license = "MIT"
+repository = "https://github.com/analog-labs/chain-connectors"
+description = "Provider wrapper for jsonrpsee."
+
+[dependencies]
+async-trait = "0.1"
+dashmap = "5.5.0"
+ethabi = "18.0.0"
+ethers = { version = "2.0", default-features = true, features = ["ws", "abigen", "rustls"] }
+futures = "0.3"
+futures-util = "0.3"
+jsonrpsee = { version = "0.20.0", default-features = false, features = ["macros"] }
+pin-project = "1.1"
+serde = "1.0"
+serde_json = "1.0"
+thiserror = "1.0"
+tokio = { version = "1.32", features = ["rt-multi-thread", "macros"] }
+tracing = "0.1"

chains/ethereum/rpc-client/src/client.rs

@@ -0,0 +1,126 @@
+use crate::{error::Error, params::RpcParams};
+use async_trait::async_trait;
+use ethers::providers::JsonRpcClient;
+use jsonrpsee::core::{
+    client::{BatchResponse, ClientT},
+    params::BatchRequestBuilder,
+    traits::ToRpcParams,
+    Error as JsonRpseeError,
+};
+use serde::de::DeserializeOwned;
+use serde::Serialize;
+use std::fmt::{Debug, Formatter};
+use std::ops::{Deref, DerefMut};
+
+// Adapter Client for JsonRpcClient trait
+#[repr(transparent)]
+pub struct EthClientAdapter<C> {
+    pub(crate) client: C,
+}
+
+impl<C> EthClientAdapter<C>
+where
+    C: ClientT + Debug + Send + Sync,
+{
+    pub fn new(client: C) -> Self {
+        Self { client }
+    }
+}
+
+impl<C> Debug for EthClientAdapter<C>
+where
+    C: Debug,
+{
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("ClientAdapter")
+            .field("client", &self.client)
+            .finish()
+    }
+}
+
+impl<C> Clone for EthClientAdapter<C>
+where
+    C: Clone,
+{
+    fn clone(&self) -> Self {
+        Self {
+            client: self.client.clone(),
+        }
+    }
+}
+
+impl<C> AsMut<C> for EthClientAdapter<C> {
+    fn as_mut(&mut self) -> &mut C {
+        &mut self.client
+    }
+}
+
+impl<C> AsRef<C> for EthClientAdapter<C> {
+    fn as_ref(&self) -> &C {
+        &self.client
+    }
+}
+
+impl<C> Deref for EthClientAdapter<C> {
+    type Target = C;
+
+    fn deref(&self) -> &Self::Target {
+        &self.client
+    }
+}
+
+impl<C> DerefMut for EthClientAdapter<C> {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.client
+    }
+}
+
+#[async_trait]
+impl<C> JsonRpcClient for EthClientAdapter<C>
+where
+    C: ClientT + Debug + Send + Sync,
+{
+    type Error = Error;
+
+    async fn request<T, R>(&self, method: &str, params: T) -> Result<R, Self::Error>
+    where
+        T: Debug + Serialize + Send + Sync,
+        R: DeserializeOwned + Send,
+    {
+        let params = RpcParams::from_serializable(&params)?;
+        ClientT::request::<R, RpcParams>(&self.client, method, params)
+            .await
+            .map_err(Error::from)
+    }
+}
+
+#[async_trait]
+impl<C> ClientT for EthClientAdapter<C>
+where
+    C: ClientT + Debug + Send + Sync,
+{
+    async fn notification<Params>(&self, method: &str, params: Params) -> Result<(), JsonRpseeError>
+    where
+        Params: ToRpcParams + Send,
+    {
+        ClientT::notification(self, method, params).await
+    }
+
+    async fn request<R, Params>(&self, method: &str, params: Params) -> Result<R, JsonRpseeError>
+    where
+        R: DeserializeOwned,
+        Params: ToRpcParams + Send,
+    {
+        ClientT::request(self, method, params).await
+    }
+
+    async fn batch_request<'a, R>(
+        &self,
+        batch: BatchRequestBuilder<'a>,
+    ) -> Result<BatchResponse<'a, R>, JsonRpseeError>
+    where
+        R: DeserializeOwned + Debug + 'a,
+    {
+        ClientT::batch_request(self, batch).await
+    }
+}

chains/ethereum/rpc-client/src/error.rs

@@ -0,0 +1,99 @@
+use ethers::providers::{
+    JsonRpcError as EthJsonRpcError, ProviderError as EthProviderError,
+    RpcError as EthRpcErrorTrait,
+};
+use jsonrpsee::{client_transport::ws::WsHandshakeError, core::Error as JsonRpseeError};
+
+#[derive(Debug, thiserror::Error)]
+pub enum Error {
+    /// Thrown if the response could not be parsed
+    #[error("{original}")]
+    JsonRpsee {
+        original: JsonRpseeError,
+        message: Option<EthJsonRpcError>,
+    },
+
+    /// Failed to parse the data.
+    #[allow(clippy::enum_variant_names)]
+    #[error(transparent)]
+    ParseError(#[from] serde_json::Error),
+
+    /// Error that can happen during the WebSocket handshake.
+    #[error("WS Handshake failed: {0}")]
+    HandshakeFailed(WsHandshakeError),
+
+    /// The background task has been terminated.
+    #[error("The background task been terminated because: {0}; restart required")]
+    RestartNeeded(String),
+
+    /// The client is reconnecting
+    #[error("The client is restarting the background task")]
+    Reconnecting,
+}
+
+impl From<JsonRpseeError> for Error {
+    fn from(error: JsonRpseeError) -> Self {
+        match error {
+            JsonRpseeError::Call(call) => {
+                let code = call.code() as i64;
+                let data = call
+                    .data()
+                    .and_then(|raw_value| serde_json::value::to_value(raw_value).ok());
+                let message = call.message().to_string();
+                Self::JsonRpsee {
+                    original: JsonRpseeError::Call(call),
+                    message: Some(EthJsonRpcError {
+                        code,
+                        message,
+                        data,
+                    }),
+                }
+            }
+            JsonRpseeError::ParseError(serde_error) => Self::ParseError(serde_error),
+            JsonRpseeError::RestartNeeded(reason) => Self::RestartNeeded(reason),
+            error => {
+                let message = format!("{}", &error);
+                Self::JsonRpsee {
+                    original: error,
+                    message: Some(EthJsonRpcError {
+                        code: 9999,
+                        message,
+                        data: None,
+                    }),
+                }
+            }
+        }
+    }
+}
+
+impl From<WsHandshakeError> for Error {
+    fn from(error: WsHandshakeError) -> Self {
+        Self::HandshakeFailed(error)
+    }
+}
+
+impl From<Error> for EthProviderError {
+    fn from(error: Error) -> Self {
+        match error {
+            Error::ParseError(error) => Self::SerdeJson(error),
+            Error::HandshakeFailed(error) => Self::CustomError(error.to_string()),
+            error => Self::JsonRpcClientError(Box::new(error)),
+        }
+    }
+}
+
+impl EthRpcErrorTrait for Error {
+    fn as_error_response(&self) -> Option<&EthJsonRpcError> {
+        match self {
+            Self::JsonRpsee { message, .. } => message.as_ref(),
+            _ => None,
+        }
+    }
+
+    fn as_serde_error(&self) -> Option<&serde_json::Error> {
+        match self {
+            Self::ParseError(error) => Some(error),
+            _ => None,
+        }
+    }
+}

chains/ethereum/rpc-client/src/lib.rs

@@ -0,0 +1,18 @@
+mod client;
+mod error;
+mod params;
+mod pubsub;
+mod subscription;
+
+pub use client::EthClientAdapter;
+pub use pubsub::EthPubsubAdapter;
+pub use subscription::SubscriptionStream;
+
+pub mod prelude {
+    pub use ethers::providers::{JsonRpcClient, PubsubClient, RpcError};
+    pub use jsonrpsee::core::{
+        client::{ClientT, SubscriptionClientT},
+        traits::ToRpcParams,
+    };
+    pub use serde::{de::DeserializeOwned, Deserialize, Serialize};
+}