refactor(batcher): retry function

batcher/aligned-batcher/Cargo.toml

@@ -19,10 +19,7 @@ bytes = "1.7.1"
 hex = "0.4.3"
 dotenvy = "0.15.0"
 anyhow = "1.0.83"
-ethers = { tag = "v2.0.15-fix-reconnections", features = [
-    "ws",
-    "rustls",
-], git = "https://github.com/yetanotherco/ethers-rs.git" }
+ethers = { version = "2.0", features = ["ws", "rustls"] }
 lambdaworks-crypto = { git = "https://github.com/lambdaclass/lambdaworks.git", rev = "efd46f0b0aea3aa95d94bba7de86cb96611b40d3", features = ["serde"] }
 serde_yaml = "0.9.34"
 sp1-sdk = { git = "https://github.com/succinctlabs/sp1.git", rev = "v1.0.1" }
@@ -35,3 +32,4 @@ priority-queue = "2.1.0"
 once_cell = "1.20.2"
 warp = "0.3.7"
 prometheus = { version = "0.13.4", features = ["process"] }
+backon = "1.2.0"

batcher/aligned-batcher/src/config/mod.rs

@@ -40,7 +40,6 @@ pub struct BatcherConfigFromYaml {
     pub block_interval: u64,
     pub max_proof_size: usize,
     pub max_batch_size: usize,
-    pub eth_ws_reconnects: usize,
     pub pre_verification_is_enabled: bool,
     pub metrics_port: u16,
     pub non_paying: Option<NonPayingConfigFromYaml>,

batcher/aligned-batcher/src/connection.rs

@@ -1,5 +1,9 @@
 use std::sync::Arc;
 
+use crate::{
+    retry::{retry_function, RetryError, DEFAULT_FACTOR, DEFAULT_MAX_TIMES, DEFAULT_MIN_DELAY},
+    types::{batch_queue::BatchQueueEntry, errors::BatcherError},
+};
 use aligned_sdk::{
     communication::serialization::cbor_serialize,
     core::types::{BatchInclusionData, ResponseMessage, VerificationCommitmentBatch},
@@ -14,8 +18,6 @@ use tokio_tungstenite::{
     WebSocketStream,
 };
 
-use crate::types::{batch_queue::BatchQueueEntry, errors::BatcherError};
-
 pub(crate) type WsMessageSink = Arc<RwLock<SplitSink<WebSocketStream<TcpStream>, Message>>>;
 
 pub(crate) async fn send_batch_inclusion_data_responses(
@@ -33,17 +35,7 @@ pub(crate) async fn send_batch_inclusion_data_responses(
             return Err(BatcherError::WsSinkEmpty);
         };
 
-        let sending_result = ws_sink
-            .write()
-            .await
-            .send(Message::binary(serialized_response))
-            .await;
-
-        match sending_result {
-            Err(Error::AlreadyClosed) => (),
-            Err(e) => error!("Error while sending batch inclusion data response: {}", e),
-            Ok(_) => (),
-        }
+        send_response_with_retry(ws_sink, serialized_response).await;
 
         info!("Response sent");
     }
@@ -66,3 +58,36 @@ pub(crate) async fn send_message<T: Serialize>(ws_conn_sink: WsMessageSink, mess
         Err(e) => error!("Error while serializing message: {}", e),
     }
 }
+
+async fn send_response_with_retry(
+    ws_sink: &Arc<RwLock<SplitSink<WebSocketStream<TcpStream>, Message>>>,
+    serialized_response: Vec<u8>,
+) {
+    if let Err(e) = retry_function(
+        || send_response(ws_sink, serialized_response.clone()),
+        DEFAULT_MIN_DELAY,
+        DEFAULT_FACTOR,
+        DEFAULT_MAX_TIMES,
+    )
+    .await
+    {
+        error!("Error while sending batch inclusion data response: {}", e);
+    }
+}
+
+async fn send_response(
+    ws_sink: &Arc<RwLock<SplitSink<WebSocketStream<TcpStream>, Message>>>,
+    serialized_response: Vec<u8>,
+) -> Result<(), RetryError<Error>> {
+    let sending_result = ws_sink
+        .write()
+        .await
+        .send(Message::binary(serialized_response))
+        .await;
+
+    match sending_result {
+        Err(Error::AlreadyClosed) => Err(RetryError::Permanent(Error::AlreadyClosed)),
+        Err(_) => Err(RetryError::Transient),
+        Ok(_) => Ok(()),
+    }
+}

batcher/aligned-batcher/src/eth/mod.rs

@@ -15,12 +15,10 @@ pub struct BatchVerified {
 }
 
 pub type SignerMiddlewareT =
-    SignerMiddleware<GasEscalatorMiddleware<Provider<RetryClient<Http>>>, Wallet<SigningKey>>;
+    SignerMiddleware<GasEscalatorMiddleware<Provider<Http>>, Wallet<SigningKey>>;
 
 pub type BatcherPaymentService = BatcherPaymentServiceContract<SignerMiddlewareT>;
 
-const MAX_RETRIES: u32 = 15; // Max retries for the retry client. Will only retry on network errors
-const INITIAL_BACKOFF: u64 = 1000; // Initial backoff for the retry client in milliseconds, will increase every retry
 const GAS_MULTIPLIER: f64 = 1.125; // Multiplier for the gas price for gas escalator
 const GAS_ESCALATOR_INTERVAL: u64 = 12; // Time in seconds between gas escalations
 
@@ -48,22 +46,15 @@ impl CreateNewTaskFeeParams {
     }
 }
 
-pub fn get_provider(eth_rpc_url: String) -> Result<Provider<RetryClient<Http>>, anyhow::Error> {
+pub fn get_provider(eth_rpc_url: String) -> Result<Provider<Http>, anyhow::Error> {
     let provider = Http::from_str(eth_rpc_url.as_str())
         .map_err(|e| anyhow::Error::msg(format!("Failed to create provider: {}", e)))?;
 
-    let client = RetryClient::new(
-        provider,
-        Box::<ethers::providers::HttpRateLimitRetryPolicy>::default(),
-        MAX_RETRIES,
-        INITIAL_BACKOFF,
-    );
-
-    Ok(Provider::<RetryClient<Http>>::new(client))
+    Ok(Provider::new(provider))
 }
 
 pub async fn get_batcher_payment_service(
-    provider: Provider<RetryClient<Http>>,
+    provider: Provider<Http>,
     ecdsa_config: ECDSAConfig,
     contract_address: String,
 ) -> Result<BatcherPaymentService, anyhow::Error> {