Migrate Sui to gRPC

#1126

Author: gemcoder21

Cargo.lock

@@ -110,6 +110,14 @@ curve25519-dalek = { version = "4.1.3" }
 ed25519-dalek = { version = "2", features = ["std"] }
 borsh = { version = "1.6.0", features = ["derive"] }
 bcs = { version = "0.2.1" }
+sui-rpc = { version = "0.3.1" }
+prost = { version = "0.14.3" }
+prost-types = { version = "0.14.3" }
+tonic = { version = "0.14.5", default-features = false, features = [
+    "channel",
+    "tls-ring",
+    "tls-webpki-roots",
+] }
 pem-rfc7468 = { version = "1.0.0", features = ["std"] }
 sui-types = { package = "sui-sdk-types", version = "0.3.0", features = [
     "serde",

Cargo.toml

@@ -77,7 +77,7 @@ name:
   did:
     url: https://indexer-basic.did.id
   suins:
-    url: https://suins-rpc.mainnet.sui.io
+    url: https://fullnode.mainnet.sui.io
   aptos:
     url: https://aptosnames.com
   injective:
@@ -132,7 +132,7 @@ chains:
   aptos:
     url: https://fullnode.mainnet.aptoslabs.com
   sui:
-    url: https://suins-rpc.mainnet.sui.io
+    url: https://fullnode.mainnet.sui.io
   celestia:
     url: https://celestia-api.polkachu.com
   injective:

Settings.yaml

@@ -59,7 +59,7 @@ chains:
   # Sui
   - chain: sui
     urls:
-      - url: https://sui-mainnet-endpoint.blockvision.org
+      - url: https://fullnode.mainnet.sui.io
 
   # XRP
   - chain: xrp

apps/dynode/chains.yml

@@ -24,9 +24,11 @@ request:
 
 headers:
   forward:
+    - accept
     - content-type
     - content-encoding
     - content-length
+    - te
   domains:
     public-eth.nownodes.io:
       - user-agent
@@ -67,11 +69,6 @@ cache:
     solana:
       - rpc_method: getVoteAccounts
         ttl: 1h
-    sui:
-      - rpc_method: suix_getValidatorsApy
-        ttl: 1h
-      - rpc_method: suix_getLatestSuiSystemState
-        ttl: 1m
     cosmos:
       - path: /cosmos/staking/v1beta1/validators
         method: GET

apps/dynode/config.yml

@@ -262,7 +262,7 @@ mod tests {
         let m = create_test_metrics();
 
         assert_eq!(m.truncate_method("eth_getBlockByNumber"), "eth_getBlockByNumber");
-        assert_eq!(m.truncate_method("suix_getLatestSuiSystemState"), "suix_getLatestSuiSystemState");
+        assert_eq!(m.truncate_method("eth_getBalance"), "eth_getBalance");
         assert_eq!(m.truncate_method("/api/v1/blocks/by_height/12345"), "/api/v1/blocks/by_height/:number");
         assert_eq!(m.truncate_method("/v1/verylongsegmentthatisgreaterthan20characters"), "/v1/:value");
     }

apps/dynode/src/metrics/mod.rs

@@ -48,8 +48,7 @@ impl DynodeBroadcastWebhookClient {
     }
 
     fn extract_payload(&self, request: &ProxyRequest, response_body: &[u8], broadcast_providers: &BroadcastProviders) -> Option<TransactionId> {
-        let response = parse_response_body(request, response_body)?;
-        let identifier = broadcast_providers.decode_transaction_broadcast(request.chain, response)?;
+        let identifier = broadcast_providers.decode_transaction_broadcast(request.chain, response_body)?;
         Some(TransactionId::new(request.chain, identifier))
     }
 
@@ -92,16 +91,6 @@ fn is_success_status(status: u16) -> bool {
     (200..300).contains(&status)
 }
 
-fn parse_response_body<'a>(request: &ProxyRequest, response_body: &'a [u8]) -> Option<&'a str> {
-    match std::str::from_utf8(response_body) {
-        Ok(response) => Some(response),
-        Err(err) => {
-            error_with_fields!("broadcast webhook decode failed", &err, chain = request.chain.as_ref(), request_id = request.id.as_str(),);
-            None
-        }
-    }
-}
-
 fn is_broadcast_request(request: &ProxyRequest, broadcast_providers: &BroadcastProviders) -> bool {
     let chain_request = match request.request_type() {
         RequestType::JsonRpc(JsonRpcRequest::Single(call)) => ChainRequest::new(ChainRequestProtocol::JsonRpc, call.method.as_str(), request.path.as_str(), &request.body),

apps/dynode/src/webhook.rs

@@ -1,4 +1,4 @@
-use std::error::Error;
+use std::{error::Error, str};
 
 use async_trait::async_trait;
 use primitives::chart::ChartCandleStick;
@@ -111,6 +111,10 @@ pub trait ChainTransactionDecode: Send + Sync {
     fn decode_transaction_broadcast(&self, _response: &str) -> Option<String> {
         None
     }
+
+    fn decode_transaction_broadcast_bytes(&self, response: &[u8]) -> Option<String> {
+        str::from_utf8(response).ok().and_then(|response| self.decode_transaction_broadcast(response))
+    }
 }
 
 #[async_trait]

crates/chain_traits/src/lib.rs

@@ -0,0 +1,64 @@
+use async_trait::async_trait;
+use std::{collections::HashMap, error::Error, fmt, sync::Arc};
+
+use crate::{
+    alien,
+    rpc::{HttpMethod, Target},
+};
+
+#[async_trait]
+pub trait GrpcTransport: Send + Sync + fmt::Debug {
+    async fn unary(&self, endpoint: &str, path: &str, body: Vec<u8>) -> Result<Vec<u8>, Box<dyn Error + Send + Sync>>;
+}
+
+fn unary_target(endpoint: &str, path: &str, body: Vec<u8>) -> Target {
+    Target {
+        url: format!("{}{}", endpoint.trim_end_matches('/'), path),
+        method: HttpMethod::Post,
+        headers: Some(grpc_headers()),
+        body: Some(body),
+    }
+}
+
+fn ensure_success_status(status: Option<u16>) -> Result<(), Box<dyn Error + Send + Sync>> {
+    if let Some(status) = status
+        && !(200..300).contains(&status)
+    {
+        return Err(format!("gRPC HTTP error: status {status}").into());
+    }
+    Ok(())
+}
+
+fn grpc_headers() -> HashMap<String, String> {
+    HashMap::from([
+        ("Content-Type".into(), "application/grpc+proto".into()),
+        ("Accept".into(), "application/grpc+proto".into()),
+        ("TE".into(), "trailers".into()),
+    ])
+}
+
+#[derive(Clone)]
+pub struct AlienGrpcTransport {
+    provider: Arc<dyn alien::RpcProvider>,
+}
+
+impl AlienGrpcTransport {
+    pub fn new(provider: Arc<dyn alien::RpcProvider>) -> Self {
+        Self { provider }
+    }
+}
+
+impl fmt::Debug for AlienGrpcTransport {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("AlienGrpcTransport").finish_non_exhaustive()
+    }
+}
+
+#[async_trait]
+impl GrpcTransport for AlienGrpcTransport {
+    async fn unary(&self, endpoint: &str, path: &str, body: Vec<u8>) -> Result<Vec<u8>, Box<dyn Error + Send + Sync>> {
+        let response = self.provider.request(unary_target(endpoint, path, body)).await?;
+        ensure_success_status(response.status)?;
+        Ok(response.data)
+    }
+}

crates/gem_jsonrpc/src/grpc.rs

@@ -5,6 +5,8 @@ pub mod alien;
 #[cfg(feature = "client")]
 pub mod client;
 #[cfg(feature = "client")]
+pub mod grpc;
+#[cfg(feature = "client")]
 pub use client::*;
 #[cfg(feature = "testkit")]
 pub mod testkit;