diff --git a/core/node/da_clients/src/eigen/client.rs b/core/node/da_clients/src/eigen/client.rs index ec9f95589a7..272217fdc5a 100644 --- a/core/node/da_clients/src/eigen/client.rs +++ b/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 { diff --git a/core/node/da_clients/src/eigen/sdk.rs b/core/node/da_clients/src/eigen/sdk.rs index 46063dd6c5e..4450ded1d33 100644 --- a/core/node/da_clients/src/eigen/sdk.rs +++ b/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)] diff --git a/eigenda-integration.md b/eigenda-integration.md index 99d86e5f473..6209996977b 100644 --- a/eigenda-integration.md +++ b/eigenda-integration.md @@ -40,8 +40,8 @@ da_client: eigenda_eth_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 ```