[bridge 23/n] wire things up (#15613)

## Description 

This PR wires things up.


## Test Plan 

Ran manual e2e test locally 

---
If your changes are not user-facing and do not break anything, you can
skip the following section. Otherwise, please briefly describe what has
changed under the Release Notes section.

### Type of Change (Check all that apply)

- [ ] protocol change
- [ ] user-visible impact
- [ ] breaking change for a client SDKs
- [ ] breaking change for FNs (FN binary must upgrade)
- [ ] breaking change for validators or node operators (must upgrade
binaries)
- [ ] breaking change for on-chain data layout
- [ ] necessitate either a data wipe or data migration

### Release notes

crates/sui-bridge/src/abi.rs

@@ -52,19 +52,22 @@ impl EthBridgeEvent {
             EthBridgeEvent::EthSuiBridgeEvents(event) => {
                 match event {
                     EthSuiBridgeEvents::TokensBridgedToSuiFilter(event) => {
-                        let Ok(event) = EthToSuiTokenBridgeV1::try_from(&event) else {
+                        let bridge_event = match EthToSuiTokenBridgeV1::try_from(&event) {
+                            Ok(bridge_event) => bridge_event,
                             // This only happens when solidity code does not align with rust code.
                             // When this happens in production, there is a risk of stuck bridge transfers.
                             // We log error here.
                             // TODO: add metrics and alert
-                            tracing::error!("Failed to convert TokensBridgedToSui log to EthToSuiTokenBridgeV1. This indicates a bug in the code: {:?}", event);
-                            return None;
+                            Err(e) => {
+                                tracing::error!(?eth_tx_hash, eth_event_index, "Failed to convert TokensBridgedToSui log to EthToSuiTokenBridgeV1. This indicates incorrect parameters or a bug in the code: {:?}. Err: {:?}", event, e);
+                                return None;
+                            }
                         };
 
                         Some(BridgeAction::EthToSuiBridgeAction(EthToSuiBridgeAction {
                             eth_tx_hash,
                             eth_event_index,
-                            eth_bridge_event: event,
+                            eth_bridge_event: bridge_event,
                         }))
                     }
                     _ => None,

crates/sui-bridge/src/client/bridge_authority_aggregator.rs

@@ -195,7 +195,7 @@ async fn request_sign_bridge_action_into_certification(
                         }
                     }
                     Err(e) => {
-                        info!(
+                        warn!(
                             "Failed to get signature from {:?}. Error: {:?}",
                             name.concise(),
                             e

crates/sui-bridge/src/client/bridge_client.rs

@@ -3,13 +3,13 @@
 
 //! `BridgeClient` talks to BridgeNode.
 
-use std::str::FromStr;
-use std::sync::Arc;
-
 use crate::crypto::{verify_signed_bridge_action, BridgeAuthorityPublicKeyBytes};
 use crate::error::{BridgeError, BridgeResult};
 use crate::server::APPLICATION_JSON;
 use crate::types::{BridgeAction, BridgeCommittee, VerifiedSignedBridgeAction};
+use fastcrypto::encoding::{Encoding, Hex};
+use std::str::FromStr;
+use std::sync::Arc;
 use url::Url;
 
 // Note: `base_url` is `Option<Url>` because `quorum_map_then_reduce_with_timeout_and_prefs`
@@ -55,8 +55,11 @@ impl BridgeClient {
                 "sign/bridge_tx/sui/eth/{}/{}",
                 e.sui_tx_digest, e.sui_tx_event_index
             ),
-            // TODO add other events
-            _ => unimplemented!(),
+            BridgeAction::EthToSuiBridgeAction(e) => format!(
+                "sign/bridge_tx/eth/sui/{}/{}",
+                Hex::encode(e.eth_tx_hash.0),
+                e.eth_event_index
+            ),
         }
     }
 
@@ -97,9 +100,11 @@ impl BridgeClient {
             .send()
             .await?;
         if !resp.status().is_success() {
+            let error_status = format!("{:?}", resp.error_for_status_ref());
             return Err(BridgeError::RestAPIError(format!(
-                "request_sign_bridge_action failed with status: {:?}",
-                resp.error_for_status()
+                "request_sign_bridge_action failed with status {:?}: {:?}",
+                error_status,
+                resp.text().await?
             )));
         }
         let signed_bridge_action = resp.json().await?;
@@ -115,18 +120,21 @@ impl BridgeClient {
 #[cfg(test)]
 mod tests {
     use crate::{
+        abi::EthToSuiTokenBridgeV1,
         crypto::BridgeAuthoritySignInfo,
+        events::EmittedSuiToEthTokenBridgeV1,
         server::mock_handler::BridgeRequestMockHandler,
         test_utils::{get_test_authority_and_key, get_test_sui_to_eth_bridge_action},
-        types::SignedBridgeAction,
+        types::{BridgeChainId, SignedBridgeAction, TokenId},
     };
     use fastcrypto::traits::KeyPair;
     use prometheus::Registry;
 
-    use crate::test_utils::run_mock_bridge_server;
-    use sui_types::{crypto::get_key_pair, digests::TransactionDigest};
-
     use super::*;
+    use crate::test_utils::run_mock_bridge_server;
+    use ethers::types::Address as EthAddress;
+    use ethers::types::TxHash;
+    use sui_types::{base_types::SuiAddress, crypto::get_key_pair, digests::TransactionDigest};
 
     #[tokio::test]
     async fn test_bridge_client() {
@@ -292,4 +300,55 @@ mod tests {
             .unwrap_err();
         assert!(matches!(err, BridgeError::MismatchedAuthoritySigner));
     }
+
+    #[test]
+    fn test_bridge_action_path_regression_tests() {
+        let sui_tx_digest = TransactionDigest::random();
+        let sui_tx_event_index = 5;
+        let action = BridgeAction::SuiToEthBridgeAction(crate::types::SuiToEthBridgeAction {
+            sui_tx_digest,
+            sui_tx_event_index,
+            sui_bridge_event: EmittedSuiToEthTokenBridgeV1 {
+                sui_chain_id: BridgeChainId::SuiDevnet,
+                nonce: 1,
+                sui_address: SuiAddress::random_for_testing_only(),
+                eth_chain_id: BridgeChainId::EthSepolia,
+                eth_address: EthAddress::random(),
+                token_id: TokenId::USDT,
+                amount: 1,
+            },
+        });
+        assert_eq!(
+            BridgeClient::bridge_action_to_path(&action),
+            format!(
+                "sign/bridge_tx/sui/eth/{}/{}",
+                sui_tx_digest, sui_tx_event_index
+            )
+        );
+
+        let eth_tx_hash = TxHash::random();
+        let eth_event_index = 6;
+        let action = BridgeAction::EthToSuiBridgeAction(crate::types::EthToSuiBridgeAction {
+            eth_tx_hash,
+            eth_event_index,
+            eth_bridge_event: EthToSuiTokenBridgeV1 {
+                eth_chain_id: BridgeChainId::EthSepolia,
+                nonce: 1,
+                eth_address: EthAddress::random(),
+                sui_chain_id: BridgeChainId::SuiDevnet,
+                sui_address: SuiAddress::random_for_testing_only(),
+                token_id: TokenId::USDT,
+                amount: 1,
+            },
+        });
+
+        assert_eq!(
+            BridgeClient::bridge_action_to_path(&action),
+            format!(
+                "sign/bridge_tx/eth/sui/{}/{}",
+                Hex::encode(eth_tx_hash.0),
+                eth_event_index
+            )
+        );
+    }
 }

crates/sui-bridge/src/config.rs

@@ -54,7 +54,7 @@ pub struct BridgeNodeConfig {
     pub sui_bridge_modules: Option<Vec<String>>,
     /// Override the start block number for each eth address. Key must be in `eth_addresses`.
     /// When set, EthSyncer will start from this block number instead of the one in storage.
-    pub eth_addresses_start_block_number_override: Option<BTreeMap<String, u64>>,
+    pub eth_bridge_contracts_start_block_override: Option<BTreeMap<String, u64>>,
     /// Override the start transaction digest for each bridge module. Key must be in `sui_bridge_modules`.
     /// When set, SuiSyncer will start from this transaction digest instead of the one in storage.
     pub sui_bridge_modules_start_tx_override: Option<BTreeMap<String, String>>,
@@ -96,14 +96,15 @@ impl BridgeNodeConfig {
         }?;
 
         let client_sui_address = SuiAddress::from(&bridge_client_key.public());
+        info!("Bridge client sui address: {:?}", client_sui_address);
         let gas_object_id = self.bridge_client_gas_object.ok_or(anyhow!(
             "`bridge_client_gas_object` is required when `run_client` is true"
         ))?;
         let db_path = self
             .db_path
             .clone()
             .ok_or(anyhow!("`db_path` is required when `run_client` is true"))?;
-        let eth_addresses = match &self.eth_addresses {
+        let eth_bridge_contracts = match &self.eth_addresses {
             Some(addresses) => {
                 if addresses.is_empty() {
                     return Err(anyhow!(
@@ -122,13 +123,13 @@ impl BridgeNodeConfig {
                 ))
             }
         };
-        let mut eth_addresses_start_block_number_override = BTreeMap::new();
-        match &self.eth_addresses_start_block_number_override {
+        let mut eth_bridge_contracts_start_block_override = BTreeMap::new();
+        match &self.eth_bridge_contracts_start_block_override {
             Some(overrides) => {
                 for (addr, block_number) in overrides {
                     let address = EthAddress::from_str(addr)?;
-                    if eth_addresses.contains(&address) {
-                        eth_addresses_start_block_number_override.insert(address, *block_number);
+                    if eth_bridge_contracts.contains(&address) {
+                        eth_bridge_contracts_start_block_override.insert(address, *block_number);
                     } else {
                         return Err(anyhow!(
                             "Override start block number for address {:?} is not in `eth_addresses`",
@@ -198,9 +199,9 @@ impl BridgeNodeConfig {
             sui_client: sui_client.clone(),
             eth_client: eth_client.clone(),
             db_path,
-            eth_addresses,
+            eth_bridge_contracts,
             sui_bridge_modules,
-            eth_addresses_start_block_number_override,
+            eth_bridge_contracts_start_block_override,
             sui_bridge_modules_start_tx_override,
         };
 
@@ -225,9 +226,9 @@ pub struct BridgeClientConfig {
     pub sui_client: Arc<SuiClient<SuiSdkClient>>,
     pub eth_client: Arc<EthClient<ethers::providers::Http>>,
     pub db_path: PathBuf,
-    pub eth_addresses: Vec<EthAddress>,
+    pub eth_bridge_contracts: Vec<EthAddress>,
     pub sui_bridge_modules: Vec<Identifier>,
-    pub eth_addresses_start_block_number_override: BTreeMap<EthAddress, u64>,
+    pub eth_bridge_contracts_start_block_override: BTreeMap<EthAddress, u64>,
     pub sui_bridge_modules_start_tx_override: BTreeMap<Identifier, TransactionDigest>,
 }
 

crates/sui-bridge/src/eth_client.rs

@@ -57,6 +57,7 @@ where
         Ok(())
     }
 
+    // TODO: need to fix this to assert address that emits the events
     pub async fn get_finalized_bridge_action_maybe(
         &self,
         tx_hash: TxHash,

crates/sui-bridge/src/main.rs

@@ -4,14 +4,24 @@
 use clap::Parser;
 use mysten_metrics::start_prometheus_server;
 use std::{
+    collections::HashMap,
     net::{IpAddr, Ipv4Addr, SocketAddr},
     path::PathBuf,
+    sync::Arc,
+    time::Duration,
 };
 use sui_bridge::{
-    config::BridgeNodeConfig,
+    action_executor::BridgeActionExecutor,
+    client::bridge_authority_aggregator::BridgeAuthorityAggregator,
+    config::{BridgeClientConfig, BridgeNodeConfig},
+    eth_syncer::EthSyncer,
+    orchestrator::BridgeOrchestrator,
     server::{handler::BridgeRequestHandler, run_server},
+    storage::BridgeOrchestratorTables,
+    sui_syncer::SuiSyncer,
 };
 use sui_config::Config;
+use tokio::task::JoinHandle;
 use tracing::info;
 
 // TODO consolidate this with sui-node/src/main.rs, but where to put it?
@@ -60,9 +70,14 @@ async fn main() -> anyhow::Result<()> {
         .with_prom_registry(&prometheus_registry)
         .init();
 
-    let (server_config, _client_config) = config.validate().await?;
+    let (server_config, client_config) = config.validate().await?;
 
-    // TODO Start Client
+    // Start Client
+    let _handles = if let Some(client_config) = client_config {
+        start_client_components(client_config).await
+    } else {
+        Ok(vec![])
+    }?;
 
     // Start Server
     let socket_address = SocketAddr::new(
@@ -80,3 +95,112 @@ async fn main() -> anyhow::Result<()> {
     .await;
     Ok(())
 }
+
+// TODO: is there a way to clean up the overrides after it's stored in DB?
+async fn start_client_components(
+    client_config: BridgeClientConfig,
+) -> anyhow::Result<Vec<JoinHandle<()>>> {
+    let store: std::sync::Arc<BridgeOrchestratorTables> =
+        BridgeOrchestratorTables::new(&client_config.db_path.join("client"));
+    let stored_module_cursors = store
+        .get_sui_event_cursors(&client_config.sui_bridge_modules)
+        .map_err(|e| anyhow::anyhow!("Unable to get sui event cursors from storage: {e:?}"))?;
+    let mut sui_modules_to_watch = HashMap::new();
+    for (module, cursor) in client_config
+        .sui_bridge_modules
+        .iter()
+        .zip(stored_module_cursors)
+    {
+        if client_config
+            .sui_bridge_modules_start_tx_override
+            .contains_key(module)
+        {
+            sui_modules_to_watch.insert(
+                module.clone(),
+                client_config.sui_bridge_modules_start_tx_override[module],
+            );
+            info!(
+                "Overriding cursor for sui bridge module {} to {}. Stored cursor: {:?}",
+                module, client_config.sui_bridge_modules_start_tx_override[module], cursor
+            );
+        } else if let Some(cursor) = cursor {
+            sui_modules_to_watch.insert(module.clone(), cursor);
+        } else {
+            return Err(anyhow::anyhow!(
+                "No cursor found for sui bridge module {} in storage or config override",
+                module
+            ));
+        }
+    }
+
+    let stored_eth_cursors = store
+        .get_eth_event_cursors(&client_config.eth_bridge_contracts)
+        .map_err(|e| anyhow::anyhow!("Unable to get eth event cursors from storage: {e:?}"))?;
+    let mut eth_contracts_to_watch = HashMap::new();
+    for (contract, cursor) in client_config
+        .eth_bridge_contracts
+        .iter()
+        .zip(stored_eth_cursors)
+    {
+        if client_config
+            .eth_bridge_contracts_start_block_override
+            .contains_key(contract)
+        {
+            eth_contracts_to_watch.insert(
+                *contract,
+                client_config.eth_bridge_contracts_start_block_override[contract],
+            );
+            info!(
+                "Overriding cursor for eth bridge contract {} to {}. Stored cursor: {:?}",
+                contract, client_config.eth_bridge_contracts_start_block_override[contract], cursor
+            );
+        } else if let Some(cursor) = cursor {
+            eth_contracts_to_watch.insert(*contract, cursor);
+        } else {
+            return Err(anyhow::anyhow!(
+                "No cursor found for eth contract {} in storage or config override",
+                contract
+            ));
+        }
+    }
+
+    let sui_client = client_config.sui_client.clone();
+
+    let mut all_handles = vec![];
+    let (task_handles, eth_events_rx, _) =
+        EthSyncer::new(client_config.eth_client.clone(), eth_contracts_to_watch)
+            .run()
+            .await
+            .expect("Failed to start eth syncer");
+    all_handles.extend(task_handles);
+
+    let (task_handles, sui_events_rx) =
+        SuiSyncer::new(client_config.sui_client, sui_modules_to_watch)
+            .run(Duration::from_secs(2))
+            .await
+            .expect("Failed to start sui syncer");
+    all_handles.extend(task_handles);
+
+    let committee = Arc::new(
+        sui_client
+            .get_committee()
+            .await
+            .expect("Failed to get committee"),
+    );
+    let bridge_auth_agg = BridgeAuthorityAggregator::new(committee);
+
+    let bridge_action_executor = BridgeActionExecutor::new(
+        sui_client.clone(),
+        Arc::new(bridge_auth_agg),
+        store.clone(),
+        client_config.key,
+        client_config.sui_address,
+        client_config.gas_object_ref.0,
+    );
+
+    let orchestrator =
+        BridgeOrchestrator::new(sui_client, sui_events_rx, eth_events_rx, store.clone());
+
+    all_handles.extend(orchestrator.run(bridge_action_executor));
+    Ok(all_handles)
+}