feat: rpc-client retry on network error

crates/block-producer/src/runner.rs

@@ -47,8 +47,8 @@ use gw_types::{
     prelude::*,
 };
 use gw_utils::{
-    exponential_backoff::ExponentialBackoff, genesis_info::CKBGenesisInfo, liveness::Liveness,
-    local_cells::LocalCellsManager, wallet::Wallet, RollupContext,
+    genesis_info::CKBGenesisInfo, liveness::Liveness, local_cells::LocalCellsManager,
+    wallet::Wallet, ExponentialBackoff, RollupContext,
 };
 use semver::Version;
 use tentacle::service::ProtocolMeta;

crates/block-producer/src/sync_l1.rs

@@ -15,7 +15,7 @@ use gw_types::{
     packed::{NumberHash, Script},
     prelude::*,
 };
-use gw_utils::{exponential_backoff::ExponentialBackoff, liveness::Liveness};
+use gw_utils::{liveness::Liveness, ExponentialBackoff};
 use tokio::sync::Mutex;
 
 use crate::chain_updater::ChainUpdater;

crates/p2p-network/src/lib.rs

@@ -6,7 +6,7 @@ use std::{
 
 use anyhow::{Context, Result};
 use gw_config::P2PNetworkConfig;
-use gw_utils::exponential_backoff::ExponentialBackoff;
+use gw_utils::ExponentialBackoff;
 use socket2::SockRef;
 use tentacle::{
     async_trait,

crates/rpc-client/src/lib.rs

@@ -7,3 +7,5 @@ pub mod indexer_types;
 pub mod rpc_client;
 mod utils;
 pub mod withdrawal;
+
+pub use utils::ExponentialBackoff;

crates/rpc-client/src/utils.rs

@@ -1,7 +1,8 @@
 use std::time::Duration;
 
-use anyhow::{Context, Result};
+use anyhow::Result;
 use jsonrpc_utils::HttpClient;
+use rand::{thread_rng, Rng};
 use reqwest::Client;
 use tracing::{field, instrument, Span};
 
@@ -40,9 +41,94 @@ impl TracingHttpClient {
             Span::current().record("params", field::display(&params));
         }
 
-        self.inner
-            .rpc(method, params)
-            .await
-            .with_context(|| format!("rpc {method}"))
+        let mut backoff = ExponentialBackoff::new(Duration::from_secs(1));
+
+        loop {
+            match self.inner.rpc(method, params).await {
+                Ok(r) => return Ok(r),
+                Err(e) => {
+                    // Retry on reqwest errors. CKB RPCs are almost all safe to retry.
+                    if e.is::<reqwest::Error>() {
+                        let next = backoff.next_sleep();
+                        // reqwest::Error displays the whole chain, no need to use {:#}.
+                        tracing::warn!(
+                            "rpc client error, will retry in {:.2}s: {}",
+                            next.as_secs_f64(),
+                            e,
+                        );
+                        tokio::time::sleep(next).await;
+                        continue;
+                    }
+                    return Err(e.context(format!("rpc {method}")));
+                }
+            }
+        }
+    }
+}
+
+pub struct ExponentialBackoff {
+    base: Duration,
+    current_multiplier: f32,
+    multiplier: f32,
+    max_sleep: Duration,
+    jitter: bool,
+}
+
+impl ExponentialBackoff {
+    pub fn new(base: Duration) -> Self {
+        Self {
+            base,
+            current_multiplier: 1.0,
+            multiplier: 2.0,
+            max_sleep: base * 32,
+            jitter: true,
+        }
+    }
+
+    pub fn next_sleep(&mut self) -> Duration {
+        let t = self.base.mul_f32(self.current_multiplier);
+        let t = if t >= self.max_sleep {
+            self.max_sleep
+        } else {
+            self.current_multiplier *= self.multiplier;
+            t
+        };
+        if self.jitter {
+            // https://aws.amazon.com/cn/blogs/architecture/exponential-backoff-and-jitter/
+            thread_rng().gen_range(Duration::ZERO..t)
+        } else {
+            t
+        }
+    }
+
+    pub fn reset(&mut self) {
+        self.current_multiplier = 1.0;
+    }
+
+    pub fn with_multiplier(self, multiplier: f32) -> Self {
+        Self { multiplier, ..self }
+    }
+
+    pub fn with_max_sleep(self, max_sleep: Duration) -> Self {
+        Self { max_sleep, ..self }
+    }
+
+    pub fn with_jitter(self, jitter: bool) -> Self {
+        Self { jitter, ..self }
+    }
+}
+
+#[cfg(test)]
+#[test]
+fn test_backoff() {
+    let mut b =
+        ExponentialBackoff::new(Duration::from_secs(1)).with_max_sleep(Duration::from_secs(64));
+    b.next_sleep();
+    assert!(b.current_multiplier == 2.0);
+    b.next_sleep();
+    assert!(b.current_multiplier == 4.0);
+    for _ in 0..10 {
+        b.next_sleep();
     }
+    assert!(b.current_multiplier == 64.0);
 }

crates/utils/src/lib.rs

@@ -1,7 +1,6 @@
 pub mod abort_on_drop;
 mod calc_finalizing_range;
 pub mod compression;
-pub mod exponential_backoff;
 pub mod export_block;
 pub mod fee;
 pub mod gasless;
@@ -20,6 +19,7 @@ pub mod wallet;
 pub mod withdrawal;
 
 pub use calc_finalizing_range::calc_finalizing_range;
+pub use gw_rpc_client::ExponentialBackoff;
 pub use query_rollup_cell::query_rollup_cell;
 pub use rollup_context::RollupContext;
 pub use timepoint::{finalized_timepoint, global_state_finalized_timepoint};