clean up wait for finality

packages/relayer-lib/src/tx_builder/eth_to_cosmos.rs

@@ -1,7 +1,6 @@
 //! This module defines [`TxBuilder`] which is responsible for building transactions to be sent to
 //! the Cosmos SDK chain from events received from Ethereum.
 
-use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
 use alloy::{primitives::Address, providers::Provider};
@@ -34,7 +33,7 @@ use prost::Message;
 use sp1_ics07_tendermint_utils::rpc::TendermintRpcExt;
 use tendermint_rpc::Client;
 
-use crate::utils::{cosmos, wait_for_condition};
+use crate::utils::{cosmos, wait_for_condition, wait_for_condition_with_capture};
 use crate::{
     chain::{CosmosSdk, EthEureka},
     events::EurekaEventWithHeight,
@@ -179,48 +178,32 @@ where
         &self,
         target_block_number: u64,
     ) -> Result<LightClientFinalityUpdate> {
-        // Shared mutable variable to store the latest update.
-        let latest_update: Arc<Mutex<Option<LightClientFinalityUpdate>>> =
-            Arc::new(Mutex::new(None));
-
-        // Clone the Arc so the closure can access it.
-        let latest_update_clone = latest_update.clone();
-        wait_for_condition(
+        // Wait until we find a finality update that meets our criteria and capture it
+        // This way we avoid making an extra call at the end
+        wait_for_condition_with_capture(
             Duration::from_secs(45 * 60),
             Duration::from_secs(10),
-            || {
-                // Clone again for the async move closure.
-                let latest_update_inner = latest_update_clone.clone();
-                async move {
-                    tracing::debug!(
-                        "Waiting for finality beyond target block number: {}",
+            || async {
+                tracing::debug!(
+                    "Waiting for finality beyond target block number: {}",
+                    target_block_number
+                );
+
+                let finality_update = self.beacon_api_client.finality_update().await?.data;
+                if finality_update.finalized_header.execution.block_number < target_block_number {
+                    tracing::info!(
+                        "Finality not found: current finality execution block number: {}, Target execution block number: {}",
+                        finality_update.finalized_header.execution.block_number,
                         target_block_number
                     );
-
-                    let finality_update = self.beacon_api_client.finality_update().await?.data;
-                    if finality_update.finalized_header.execution.block_number < target_block_number
-                    {
-                        tracing::info!(
-                            "Finality not found: current finality execution block number: {}, Target execution block number: {}",
-                            finality_update.finalized_header.execution.block_number,
-                            target_block_number
-                        );
-                        return Ok(false);
-                    }
-
-                    // Store the current update.
-                    *latest_update_inner.lock().unwrap() = Some(finality_update);
-                    Ok(true)
+                    return Ok(None);
                 }
+
+                // Return the update itself when the condition is met
+                Ok(Some(finality_update))
             },
         )
-        .await?;
-
-        // Retrieve the stored update.
-        let stored_update = latest_update.lock().unwrap().take().ok_or_else(|| {
-            anyhow::anyhow!("Finality update was not stored even though condition was met")
-        })?;
-        Ok(stored_update)
+        .await
     }
 
     async fn light_client_update_to_header(

packages/relayer-lib/src/utils/mod.rs

@@ -5,6 +5,8 @@ use std::future::Future;
 use std::time::{Duration, Instant};
 
 /// Retries an operation until the condition is met or a timeout occurs.
+///
+/// The basic version just checks for a boolean condition.
 pub async fn wait_for_condition<F, Fut>(
     timeout: Duration,
     interval: Duration,
@@ -29,5 +31,32 @@ where
     anyhow::bail!("Timeout exceeded")
 }
 
+/// Retries an operation until the condition is met or a timeout occurs, with a value capture.
+/// Returns the value that was captured when the condition was met.
+pub async fn wait_for_condition_with_capture<F, Fut, T>(
+    timeout: Duration,
+    interval: Duration,
+    mut condition: F,
+) -> anyhow::Result<T>
+where
+    F: FnMut() -> Fut + Send,
+    Fut: Future<Output = anyhow::Result<Option<T>>> + Send,
+    T: Send + 'static,
+{
+    let start = Instant::now();
+    while start.elapsed() < timeout {
+        if let Some(value) = condition().await? {
+            return Ok(value);
+        }
+
+        tracing::debug!(
+            "Condition not met. Waiting for {} seconds before retrying",
+            interval.as_secs()
+        );
+        Delay::new(interval).await;
+    }
+    anyhow::bail!("Timeout exceeded")
+}
+
 pub mod cosmos;
 pub mod eth_eureka;

programs/relayer/src/modules/eth_to_cosmos.rs

@@ -7,7 +7,7 @@ use alloy::{
     providers::{Provider, RootProvider},
 };
 use ibc_eureka_relayer_lib::{
-    events::EurekaEvent,
+    events::{EurekaEvent, EurekaEventWithHeight},
     listener::{cosmos_sdk, eth_eureka, ChainListenerService},
     tx_builder::{eth_to_cosmos, TxBuilderService},
 };
@@ -36,7 +36,7 @@ struct EthToCosmosRelayerModuleService {
 }
 
 enum EthToCosmosTxBuilder {
-    Real(eth_to_cosmos::TxBuilder<RootProvider>),
+    Real(eth_to_cosmos::TxBuilder<RootProvider, HttpClient>),
     Mock(eth_to_cosmos::MockTxBuilder<RootProvider>),
 }
 
@@ -178,9 +178,17 @@ impl RelayerService for EthToCosmosRelayerModuleService {
             cosmos_events.len()
         );
 
+        // Extract the inner events from the EurekaEventWithHeight, as our relay_events takes Vec<EurekaEvent>
+        let eth_events_inner = eth_events.into_iter().map(|e| e.event).collect();
+        let cosmos_events_inner = cosmos_events.into_iter().map(|e| e.event).collect();
+
         let tx = self
             .tx_builder
-            .relay_events(eth_events, cosmos_events, inner_req.target_client_id)
+            .relay_events(
+                eth_events_inner,
+                cosmos_events_inner,
+                inner_req.target_client_id,
+            )
             .await
             .map_err(|e| tonic::Status::from_error(e.to_string().into()))?;
 
@@ -219,14 +227,39 @@ impl EthToCosmosTxBuilder {
         target_events: Vec<EurekaEvent>,
         target_client_id: String,
     ) -> anyhow::Result<Vec<u8>> {
+        // Convert EurekaEvent to EurekaEventWithHeight
+        let src_events_with_height = src_events
+            .into_iter()
+            .map(|event| EurekaEventWithHeight {
+                event,
+                block_number: None,
+            })
+            .collect();
+
+        let target_events_with_height = target_events
+            .into_iter()
+            .map(|event| EurekaEventWithHeight {
+                event,
+                block_number: None,
+            })
+            .collect();
+
         match self {
             Self::Real(tb) => {
-                tb.relay_events(src_events, target_events, target_client_id)
-                    .await
+                tb.relay_events(
+                    src_events_with_height,
+                    target_events_with_height,
+                    target_client_id,
+                )
+                .await
             }
             Self::Mock(tb) => {
-                tb.relay_events(src_events, target_events, target_client_id)
-                    .await
+                tb.relay_events(
+                    src_events_with_height,
+                    target_events_with_height,
+                    target_client_id,
+                )
+                .await
             }
         }
     }