feat(eigen-client-extra-features): Add soft confirmations (#322)

* initial commit

* Add tests

* Add memstore

* Add assert to tests

* Add rest of memstore

* Add soft confirmations

* Remove print

* Address pr comments

* Remove to retriable error

* Fix conflicts

* Add inclusion data

* Change status query timeout to millis

* Document memstore

* Fix typo

* Fix typo

* Format

---------

Co-authored-by: Juan Munoz <[email protected]>

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

@@ -120,8 +120,8 @@ mod tests {
             eigenda_eth_rpc: String::default(),
             eigenda_svc_manager_address: "0xD4A7E1Bd8015057293f0D0A557088c286942e84b".to_string(),
             blob_size_limit: 2 * 1024 * 1024, // 2MB
-            status_query_timeout: 1800,       // 30 minutes
-            status_query_interval: 5,         // 5 seconds
+            status_query_timeout: 1800000,    // 30 minutes
+            status_query_interval: 5,         // 5 ms
             wait_for_finalization: false,
             authenticated: false,
         });
@@ -157,8 +157,8 @@ mod tests {
             eigenda_eth_rpc: String::default(),
             eigenda_svc_manager_address: "0xD4A7E1Bd8015057293f0D0A557088c286942e84b".to_string(),
             blob_size_limit: 2 * 1024 * 1024, // 2MB
-            status_query_timeout: 1800,       // 30 minutes
-            status_query_interval: 5,         // 5 seconds
+            status_query_timeout: 1800000,    // 30 minutes
+            status_query_interval: 5,         // 5 ms
             wait_for_finalization: false,
             authenticated: true,
         });
@@ -218,6 +218,43 @@ mod tests {
         assert_eq!(retrieved_data.unwrap(), data);
     }
 
+    #[tokio::test]
+    async fn test_wait_for_finalization() {
+        let config = EigenConfig::Disperser(DisperserConfig {
+            custom_quorum_numbers: None,
+            disperser_rpc: "https://disperser-holesky.eigenda.xyz:443".to_string(),
+            eth_confirmation_depth: -1,
+            eigenda_eth_rpc: String::default(),
+            eigenda_svc_manager_address: "0xD4A7E1Bd8015057293f0D0A557088c286942e84b".to_string(),
+            blob_size_limit: 2 * 1024 * 1024, // 2MB
+            status_query_timeout: 1800000,    // 30 minutes
+            status_query_interval: 5000,      // 5000 ms
+            wait_for_finalization: true,
+            authenticated: true,
+        });
+        let secrets = EigenSecrets {
+            private_key: PrivateKey::from_str(
+                "d08aa7ae1bb5ddd46c3c2d8cdb5894ab9f54dec467233686ca42629e826ac4c6",
+            )
+            .unwrap(),
+        };
+        let client = EigenClient::new(config, secrets).await.unwrap();
+        let data = vec![1; 20];
+        let result = client.dispatch_blob(0, data.clone()).await.unwrap();
+        let blob_info: BlobInfo =
+            rlp::decode(&hex::decode(result.blob_id.clone()).unwrap()).unwrap();
+        let expected_inclusion_data = blob_info.blob_verification_proof.inclusion_proof;
+        let actual_inclusion_data = client
+            .get_inclusion_data(&result.blob_id)
+            .await
+            .unwrap()
+            .unwrap()
+            .data;
+        assert_eq!(expected_inclusion_data, actual_inclusion_data);
+        let retrieved_data = client.get_blob_data(&result.blob_id).await.unwrap();
+        assert_eq!(retrieved_data.unwrap(), data);
+    }
+
     #[tokio::test]
     async fn test_eigenda_dispatch_blob_too_large() {
         let config = EigenConfig::MemStore(MemStoreConfig {

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

@@ -1,7 +1,7 @@
 use std::{str::FromStr, time::Duration};
 
 use secp256k1::{ecdsa::RecoverableSignature, SecretKey};
-use tokio::sync::mpsc;
+use tokio::{sync::mpsc, time::Instant};
 use tokio_stream::{wrappers::ReceiverStream, StreamExt};
 use tonic::{
     transport::{Channel, ClientTlsConfig, Endpoint},
@@ -221,7 +221,9 @@ impl RawEigenClient {
             request_id: disperse_blob_reply.request_id,
         };
 
-        loop {
+        let start_time = Instant::now();
+        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
                 .get_blob_status(polling_request.clone())
@@ -236,7 +238,15 @@ impl RawEigenClient {
                 disperser::BlobStatus::InsufficientSignatures => {
                     return Err(anyhow::anyhow!("Insufficient signatures"))
                 }
-                disperser::BlobStatus::Confirmed | disperser::BlobStatus::Finalized => {
+                disperser::BlobStatus::Confirmed => {
+                    if !self.config.wait_for_finalization {
+                        let blob_info = resp
+                            .info
+                            .ok_or_else(|| anyhow::anyhow!("No blob header in response"))?;
+                        return Ok(blob_info);
+                    }
+                }
+                disperser::BlobStatus::Finalized => {
                     let blob_info = resp
                         .info
                         .ok_or_else(|| anyhow::anyhow!("No blob header in response"))?;
@@ -246,6 +256,8 @@ impl RawEigenClient {
                 _ => return Err(anyhow::anyhow!("Received unknown blob status")),
             }
         }
+
+        Err(anyhow::anyhow!("Failed to disperse blob (timeout)"))
     }
 
     #[cfg(test)]

eigenda-integration.md

@@ -40,8 +40,8 @@ da_client:
       eigenda_eth_rpc: <your_desired_rpc>
       eigenda_svc_manager_address: '0xD4A7E1Bd8015057293f0D0A557088c286942e84b'
       blob_size_limit: 2097152
-      status_query_timeout: 1800
-      status_query_interval: 5
+      status_query_timeout: 1800000 # ms
+      status_query_interval: 5 # ms
       wait_for_finalization: false
       authenticated: false
 ```