Merge pull request #892 from quake/quake/ckb-rpc-client-timeout

chore: timeout ckb rpc call

Author: quake

Cargo.lock

@@ -20,7 +20,7 @@ bitmask-enum = "2.2.5"
 bs58 = "0.5.1"
 ckb-gen-types = "0.202.0"
 ckb-jsonrpc-types = "0.202.0"
-ckb-sdk = "4.1.0"
+ckb-sdk = "4.4"
 ckb-testtool = {version = "0.16.0", optional = true}
 ckb-types = "0.202.0"
 clap = {version = "4.5.2", features = ["derive", "env", "string"]}

crates/fiber-lib/Cargo.toml

@@ -254,8 +254,7 @@ impl Actor for CkbChainActor {
                 }
             }
             CkbChainMessage::SendTx(tx, reply_port) => {
-                let rpc_url = state.config.rpc_url.clone();
-                let ckb_client = CkbRpcAsyncClient::new(&rpc_url);
+                let ckb_client = state.config.ckb_rpc_client();
                 let result = match ckb_client.send_transaction(tx.data().into(), None).await {
                     Ok(_) => Ok(()),
                     Err(err) => {
@@ -289,8 +288,7 @@ impl Actor for CkbChainActor {
                 }
             }
             CkbChainMessage::GetTx(tx_hash, reply_port) => {
-                let rpc_url = state.config.rpc_url.clone();
-                let ckb_client = CkbRpcAsyncClient::new(&rpc_url);
+                let ckb_client = state.config.ckb_rpc_client();
                 let result = ckb_client.get_transaction(tx_hash.into()).await;
                 if !reply_port.is_closed() {
                     // ignore error
@@ -317,8 +315,7 @@ impl Actor for CkbChainActor {
                 GetBlockTimestampRequest { block_hash },
                 reply_port,
             ) => {
-                let rpc_url = state.config.rpc_url.clone();
-                let ckb_client = CkbRpcAsyncClient::new(&rpc_url);
+                let ckb_client = state.config.ckb_rpc_client();
                 let _ = reply_port.send(
                     ckb_client
                         .get_header(block_hash.into())
@@ -327,8 +324,7 @@ impl Actor for CkbChainActor {
                 );
             }
             CkbChainMessage::GetShutdownTx(request, reply_port) => {
-                let rpc_url = state.config.rpc_url.clone();
-                let client = CkbRpcAsyncClient::new(&rpc_url);
+                let client = state.config.ckb_rpc_client();
                 let response = get_shutdown_tx(&client, request).await;
                 let _ = reply_port.send(response);
             }

crates/fiber-lib/src/ckb/actor.rs

@@ -3,6 +3,7 @@ use super::contracts::{get_script_by_contract, Contract};
 use crate::utils::encrypt_decrypt_file::{decrypt_from_file, encrypt_to_file};
 use crate::Result;
 use ckb_jsonrpc_types::{OutPoint as OutPointWrapper, Script as ScriptWrapper};
+use ckb_sdk::{traits::DefaultCellCollector, CkbRpcAsyncClient};
 use ckb_types::core::ScriptHashType;
 use ckb_types::prelude::Builder;
 use ckb_types::H256;
@@ -109,6 +110,11 @@ impl CkbConfig {
             Ok(())
         }
     }
+
+    pub fn ckb_rpc_client(&self) -> CkbRpcAsyncClient {
+        new_ckb_rpc_async_client(&self.rpc_url)
+    }
+
     #[cfg(target_arch = "wasm32")]
     pub fn read_secret_key(&self) -> Result<SecretKey> {
         Ok(self.wasm_secret_key.expect("SecretKey not found on wasm"))
@@ -280,3 +286,21 @@ impl From<&UdtCellDep> for CellDep {
             .build()
     }
 }
+
+pub const CKB_RPC_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
+
+pub fn new_ckb_rpc_async_client(rpc_url: &str) -> CkbRpcAsyncClient {
+    #[cfg(not(target_arch = "wasm32"))]
+    return CkbRpcAsyncClient::with_builder(rpc_url, |builder| builder.timeout(CKB_RPC_TIMEOUT))
+        .expect("create ckb rpc client should not fail");
+    #[cfg(target_arch = "wasm32")]
+    return CkbRpcAsyncClient::new(rpc_url);
+}
+
+pub fn new_default_cell_collector(rpc_url: &str) -> DefaultCellCollector {
+    #[cfg(not(target_arch = "wasm32"))]
+    return DefaultCellCollector::new_with_timeout(rpc_url, CKB_RPC_TIMEOUT)
+        .expect("create default cell collector should not fail");
+    #[cfg(target_arch = "wasm32")]
+    return DefaultCellCollector::new(rpc_url);
+}

crates/fiber-lib/src/ckb/config.rs

@@ -1,7 +1,4 @@
-use ckb_sdk::{
-    rpc::ckb_indexer::{Order, ScriptType, SearchKey, SearchMode},
-    CkbRpcAsyncClient,
-};
+use ckb_sdk::rpc::ckb_indexer::{Order, ScriptType, SearchKey, SearchMode};
 use ckb_types::{
     core::{BlockView, DepType, ScriptHashType},
     packed::{CellDep, CellDepVec, CellDepVecBuilder, CellOutput, OutPoint, Script},
@@ -14,9 +11,12 @@ use std::{collections::HashMap, vec};
 use thiserror::Error;
 use tracing::info;
 
-use crate::fiber::{
-    config::FiberScript,
-    gen::fiber::{UdtDep, UdtDepUnion},
+use crate::{
+    ckb::config::new_ckb_rpc_async_client,
+    fiber::{
+        config::FiberScript,
+        gen::fiber::{UdtDep, UdtDepUnion},
+    },
 };
 
 use super::config::{UdtArgInfo, UdtCfgInfos};
@@ -82,7 +82,7 @@ impl TypeIDResolver {
     }
 
     pub async fn resolve(&self, type_id: Script) -> Option<CellDep> {
-        let ckb_client = CkbRpcAsyncClient::new(&self.ckb_url);
+        let ckb_client = new_ckb_rpc_async_client(&self.ckb_url);
         let search_key = SearchKey {
             script: type_id.into(),
             script_type: ScriptType::Type,

crates/fiber-lib/src/ckb/contracts.rs

@@ -1,17 +1,23 @@
 use super::super::FundingError;
-use crate::{ckb::contracts::get_udt_cell_deps, fiber::serde_utils::EntityHex};
+use crate::{
+    ckb::{
+        config::{new_ckb_rpc_async_client, new_default_cell_collector},
+        contracts::get_udt_cell_deps,
+    },
+    fiber::serde_utils::EntityHex,
+};
 use anyhow::anyhow;
 use ckb_sdk::{
     constants::SIGHASH_TYPE_HASH,
     rpc::ckb_indexer::SearchMode,
     traits::{
-        CellCollector, CellDepResolver, CellQueryOptions, DefaultCellCollector,
-        DefaultCellDepResolver, DefaultHeaderDepResolver, DefaultTransactionDependencyProvider,
-        HeaderDepResolver, SecpCkbRawKeySigner, TransactionDependencyProvider, ValueRangeOption,
+        CellCollector, CellDepResolver, CellQueryOptions, DefaultCellDepResolver,
+        DefaultHeaderDepResolver, DefaultTransactionDependencyProvider, HeaderDepResolver,
+        SecpCkbRawKeySigner, TransactionDependencyProvider, ValueRangeOption,
     },
     tx_builder::{unlock_tx_async, CapacityBalancer, TxBuilder, TxBuilderError},
     unlock::{ScriptUnlocker, SecpSighashUnlocker},
-    CkbRpcAsyncClient, ScriptId,
+    ScriptId,
 };
 use ckb_types::{
     core::{BlockView, Capacity, TransactionView},
@@ -392,7 +398,7 @@ impl FundingTxBuilder {
             self.request.funding_fee_rate,
         );
 
-        let ckb_client = CkbRpcAsyncClient::new(&self.context.rpc_url);
+        let ckb_client = new_ckb_rpc_async_client(&self.context.rpc_url);
         let cell_dep_resolver = {
             match ckb_client.get_block_by_number(0.into()).await? {
                 Some(genesis_block) => {
@@ -419,7 +425,7 @@ impl FundingTxBuilder {
         };
 
         let header_dep_resolver = DefaultHeaderDepResolver::new(&self.context.rpc_url);
-        let mut cell_collector = DefaultCellCollector::new(&self.context.rpc_url);
+        let mut cell_collector = new_default_cell_collector(&self.context.rpc_url);
         let tx_dep_provider = DefaultTransactionDependencyProvider::new(&self.context.rpc_url, 10);
 
         let tip_block_number: u64 = ckb_client.get_tip_block_number().await?.into();
@@ -621,7 +627,7 @@ impl FundingTx {
             return Err(FundingError::InvalidPeerFundingTx);
         }
         // Peer SHOULD NOT add inputs locked by our lock scripts
-        let ckb_client = CkbRpcAsyncClient::new(&context.rpc_url);
+        let ckb_client = new_ckb_rpc_async_client(&context.rpc_url);
         for input in remote_tx.input_pts_iter().skip(local_tx.inputs().len()) {
             match ckb_client.get_live_cell(input.into(), false).await?.cell {
                 Some(cell) => {

crates/fiber-lib/src/ckb/funding/funding_tx.rs

@@ -1,11 +1,10 @@
 use std::collections::HashMap;
 
 use bitmask_enum::bitmask;
-use ckb_sdk::CkbRpcAsyncClient;
 use ckb_types::core::tx_pool::TxStatus;
 use ractor::{concurrency::Duration, Actor, ActorProcessingErr, ActorRef, RpcReplyPort};
 
-use crate::fiber::types::Hash256;
+use crate::{ckb::config::new_ckb_rpc_async_client, fiber::types::Hash256};
 
 use super::jsonrpc_types_convert::tx_status_from_json;
 
@@ -266,7 +265,7 @@ impl TracingTask {
     }
 
     async fn run_inner(self) -> Result<(), Box<dyn std::error::Error>> {
-        let ckb_client = CkbRpcAsyncClient::new(&self.rpc_url);
+        let ckb_client = new_ckb_rpc_async_client(&self.rpc_url);
         let tip_block_number: u64 = ckb_client.get_tip_block_number().await?.into();
 
         for tx_hash in self.tx_hashes {

crates/fiber-lib/src/ckb/tx_tracing_actor.rs

@@ -21,6 +21,7 @@ use tracing::{debug, error, info, warn};
 
 use crate::{
     ckb::{
+        config::{new_default_cell_collector, CKB_RPC_TIMEOUT},
         contracts::{get_cell_deps_sync, get_script_by_contract, Contract},
         CkbConfig,
     },
@@ -147,10 +148,13 @@ where
         let secret_key = state.secret_key;
         let rpc_url = state.config.rpc_url.clone();
         tokio::task::block_in_place(move || {
-            let mut cell_collector = DefaultCellCollector::new(&rpc_url);
+            let mut cell_collector = new_default_cell_collector(&rpc_url);
 
             for channel_data in self.store.get_watch_channels() {
-                let ckb_client = CkbRpcClient::new(&rpc_url);
+                let ckb_client = CkbRpcClient::with_builder(&rpc_url, |builder| {
+                    builder.timeout(CKB_RPC_TIMEOUT)
+                })
+                .expect("create ckb rpc client should not fail");
                 let search_key = SearchKey {
                     script: channel_data.funding_tx_lock.clone().into(),
                     script_type: ScriptType::Lock,