feat(eigen-client-m0-implementation): optimize concurrent dispatcher

core/node/da_clients/src/eigen/sdk.rs

@@ -1,7 +1,10 @@
-use std::{str::FromStr, time::Duration};
+use std::{str::FromStr, sync::Arc, time::Duration};
 
 use secp256k1::{ecdsa::RecoverableSignature, SecretKey};
-use tokio::{sync::mpsc, time::Instant};
+use tokio::{
+    sync::{mpsc, Mutex},
+    time::Instant,
+};
 use tokio_stream::{wrappers::UnboundedReceiverStream, StreamExt};
 use tonic::{
     transport::{Channel, ClientTlsConfig, Endpoint},
@@ -28,7 +31,7 @@ use crate::eigen::{
 
 #[derive(Debug, Clone)]
 pub(crate) struct RawEigenClient {
-    client: DisperserClient<Channel>,
+    client: Arc<Mutex<DisperserClient<Channel>>>,
     private_key: SecretKey,
     pub config: DisperserConfig,
     verifier: Verifier,
@@ -40,9 +43,11 @@ impl RawEigenClient {
     pub async fn new(private_key: SecretKey, config: DisperserConfig) -> anyhow::Result<Self> {
         let endpoint =
             Endpoint::from_str(config.disperser_rpc.as_str())?.tls_config(ClientTlsConfig::new())?;
-        let client = DisperserClient::connect(endpoint)
-            .await
-            .map_err(|e| anyhow::anyhow!("Failed to connect to Disperser server: {}", e))?;
+        let client = Arc::new(Mutex::new(
+            DisperserClient::connect(endpoint)
+                .await
+                .map_err(|e| anyhow::anyhow!("Failed to connect to Disperser server: {}", e))?,
+        ));
 
         let verifier_config = VerifierConfig {
             verify_certs: true,
@@ -70,13 +75,16 @@ impl RawEigenClient {
             account_id: String::default(), // Account Id is not used in non-authenticated mode
         };
 
-        let mut client_clone = self.client.clone();
-        let disperse_reply = client_clone.disperse_blob(request).await?.into_inner();
+        let disperse_reply = self
+            .client
+            .lock()
+            .await
+            .disperse_blob(request)
+            .await?
+            .into_inner();
 
         let disperse_time = Instant::now();
-        let blob_info = self
-            .await_for_inclusion(client_clone, disperse_reply)
-            .await?;
+        let blob_info = self.await_for_inclusion(disperse_reply).await?;
         let disperse_elapsed = Instant::now() - disperse_time;
 
         let blob_info = blob_info::BlobInfo::try_from(blob_info)
@@ -116,12 +124,13 @@ impl RawEigenClient {
     }
 
     async fn dispatch_blob_authenticated(&self, data: Vec<u8>) -> anyhow::Result<String> {
-        let mut client_clone = self.client.clone();
         let (tx, rx) = mpsc::unbounded_channel();
 
         let disperse_time = Instant::now();
+        let client_clone = self.client.clone();
+        let mut client_lock = client_clone.lock().await;
         let response_stream =
-            client_clone.disperse_blob_authenticated(UnboundedReceiverStream::new(rx));
+            client_lock.disperse_blob_authenticated(UnboundedReceiverStream::new(rx));
         let padded_data = convert_by_padding_empty_byte(&data);
 
         // 1. send DisperseBlobRequest
@@ -145,15 +154,14 @@ impl RawEigenClient {
             .unwrap()
             .payload
             .ok_or_else(|| anyhow::anyhow!("No payload in response"))?;
+        drop(client_lock);
 
         let disperser::authenticated_reply::Payload::DisperseReply(disperse_reply) = reply else {
             return Err(anyhow::anyhow!("Unexpected response from server"));
         };
 
         // 5. poll for blob status until it reaches the Confirmed state
-        let blob_info = self
-            .await_for_inclusion(client_clone, disperse_reply)
-            .await?;
+        let blob_info = self.await_for_inclusion(disperse_reply).await?;
 
         let blob_info = blob_info::BlobInfo::try_from(blob_info)
             .map_err(|e| anyhow::anyhow!("Failed to convert blob info: {}", e))?;
@@ -255,7 +263,6 @@ impl RawEigenClient {
 
     async fn await_for_inclusion(
         &self,
-        mut client: DisperserClient<Channel>,
         disperse_blob_reply: DisperseBlobReply,
     ) -> anyhow::Result<DisperserBlobInfo> {
         let polling_request = disperser::BlobStatusRequest {
@@ -266,7 +273,10 @@ impl RawEigenClient {
         while Instant::now() - start_time < Duration::from_millis(self.config.status_query_timeout)
         {
             tokio::time::sleep(Duration::from_millis(self.config.status_query_interval)).await;
-            let resp = client
+            let resp = self
+                .client
+                .lock()
+                .await
                 .get_blob_status(polling_request.clone())
                 .await?
                 .into_inner();
@@ -323,7 +333,8 @@ impl RawEigenClient {
             .batch_header_hash;
         let get_response = self
             .client
-            .clone()
+            .lock()
+            .await
             .retrieve_blob(disperser::RetrieveBlobRequest {
                 batch_header_hash,
                 blob_index,