[bridge] add watchdog to bridge node (#19878)

## Description 

1. run watchdogs on bridge nodes.
2. merges `sui-bridge-watchdog` to `sui-bridge` crate, so there is no
circular dependencies

## Test plan 

unit tests, will deploy locally and test

---

## Release notes

Check each box that your changes affect. If none of the boxes relate to
your changes, release notes aren't required.

For each box you select, include information after the relevant heading
that describes the impact of your changes that a user might notice and
any actions they must take to implement updates.

- [ ] Protocol: 
- [ ] Nodes (Validators and Full nodes): 
- [ ] Indexer: 
- [ ] JSON-RPC: 
- [ ] GraphQL: 
- [ ] CLI: 
- [ ] Rust SDK:
- [ ] REST API:

Cargo.toml

@@ -93,7 +93,6 @@ members = [
     "crates/sui-bridge",
     "crates/sui-bridge-cli",
     "crates/sui-bridge-indexer",
-    "crates/sui-bridge-watchdog",
     "crates/sui-cluster-test",
     "crates/sui-config",
     "crates/sui-core",
@@ -619,7 +618,6 @@ sui-archival = { path = "crates/sui-archival" }
 sui-authority-aggregation = { path = "crates/sui-authority-aggregation" }
 sui-benchmark = { path = "crates/sui-benchmark" }
 sui-bridge = { path = "crates/sui-bridge" }
-sui-bridge-watchdog = { path = "crates/sui-bridge-watchdog" }
 sui-cluster-test = { path = "crates/sui-cluster-test" }
 sui-config = { path = "crates/sui-config" }
 sui-core = { path = "crates/sui-core" }

crates/sui-bridge-indexer/Cargo.toml

@@ -36,7 +36,6 @@ backoff.workspace = true
 sui-config.workspace = true
 tempfile.workspace = true
 sui-indexer-builder.workspace = true
-sui-bridge-watchdog.workspace = true
 
 [dev-dependencies]
 sui-types = { workspace = true, features = ["test-utils"] }

crates/sui-bridge-indexer/src/main.rs

@@ -15,6 +15,7 @@ use std::str::FromStr;
 use std::sync::Arc;
 use sui_bridge::eth_client::EthClient;
 use sui_bridge::metered_eth_provider::{new_metered_eth_provider, MeteredEthHttpProvier};
+use sui_bridge::sui_bridge_watchdog::Observable;
 use sui_bridge::sui_client::SuiBridgeClient;
 use sui_bridge::utils::get_eth_contract_addresses;
 use sui_bridge_indexer::eth_bridge_indexer::EthFinalizedSyncDatasource;
@@ -28,6 +29,10 @@ use mysten_metrics::spawn_logged_monitored_task;
 use mysten_metrics::start_prometheus_server;
 
 use sui_bridge::metrics::BridgeMetrics;
+use sui_bridge::sui_bridge_watchdog::{
+    eth_bridge_status::EthBridgeStatus, eth_vault_balance::EthVaultBalance,
+    metrics::WatchdogMetrics, sui_bridge_status::SuiBridgeStatus, BridgeWatchDog,
+};
 use sui_bridge_indexer::config::IndexerConfig;
 use sui_bridge_indexer::eth_bridge_indexer::EthDataMapper;
 use sui_bridge_indexer::metrics::BridgeIndexerMetrics;
@@ -37,10 +42,6 @@ use sui_bridge_indexer::sui_bridge_indexer::SuiBridgeDataMapper;
 use sui_bridge_indexer::sui_datasource::SuiCheckpointDatasource;
 use sui_bridge_indexer::sui_transaction_handler::handle_sui_transactions_loop;
 use sui_bridge_indexer::sui_transaction_queries::start_sui_tx_polling_task;
-use sui_bridge_watchdog::{
-    eth_bridge_status::EthBridgeStatus, eth_vault_balance::EthVaultBalance,
-    metrics::WatchdogMetrics, sui_bridge_status::SuiBridgeStatus, BridgeWatchDog,
-};
 use sui_data_ingestion_core::DataIngestionMetrics;
 use sui_indexer_builder::indexer_builder::{BackfillStrategy, IndexerBuilder};
 use sui_indexer_builder::progress::{
@@ -247,14 +248,12 @@ async fn start_watchdog(
 
     let sui_bridge_status =
         SuiBridgeStatus::new(sui_client, watchdog_metrics.sui_bridge_paused.clone());
-
-    BridgeWatchDog::new(vec![
-        Arc::new(eth_vault_balance),
-        Arc::new(eth_bridge_status),
-        Arc::new(sui_bridge_status),
-    ])
-    .run()
-    .await;
+    let observables: Vec<Box<dyn Observable + Send + Sync>> = vec![
+        Box::new(eth_vault_balance),
+        Box::new(eth_bridge_status),
+        Box::new(sui_bridge_status),
+    ];
+    BridgeWatchDog::new(observables).run().await;
     Ok(())
 }
 

crates/sui-bridge/src/config.rs

@@ -17,6 +17,7 @@ use ethers::types::Address as EthAddress;
 use futures::{future, StreamExt};
 use serde::{Deserialize, Serialize};
 use serde_with::serde_as;
+use std::collections::BTreeMap;
 use std::collections::HashSet;
 use std::path::PathBuf;
 use std::str::FromStr;
@@ -119,6 +120,9 @@ pub struct BridgeNodeConfig {
     pub metrics_key_pair: NetworkKeyPair,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub metrics: Option<MetricsConfig>,
+
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub watchdog_config: Option<WatchdogConfig>,
 }
 
 pub fn default_ed25519_key_pair() -> NetworkKeyPair {
@@ -133,6 +137,13 @@ pub struct MetricsConfig {
     pub push_url: String,
 }
 
+#[derive(Debug, Clone, Deserialize, Serialize)]
+#[serde(rename_all = "kebab-case")]
+pub struct WatchdogConfig {
+    /// Total supplies to watch on Sui. Mapping from coin name to coin type tag
+    pub total_supplies: BTreeMap<String, String>,
+}
+
 impl Config for BridgeNodeConfig {}
 
 impl BridgeNodeConfig {
@@ -197,6 +208,7 @@ impl BridgeNodeConfig {
         let bridge_server_config = BridgeServerConfig {
             key: bridge_authority_key,
             metrics_port: self.metrics_port,
+            eth_bridge_proxy_address: eth_contracts[0], // the first contract is bridge proxy
             server_listen_port: self.server_listen_port,
             sui_client: sui_client.clone(),
             eth_client: eth_client.clone(),
@@ -385,6 +397,7 @@ impl BridgeNodeConfig {
 pub struct BridgeServerConfig {
     pub key: BridgeAuthorityKeyPair,
     pub server_listen_port: u16,
+    pub eth_bridge_proxy_address: EthAddress,
     pub metrics_port: u16,
     pub sui_client: Arc<SuiClient<SuiSdkClient>>,
     pub eth_client: Arc<EthClient<MeteredEthHttpProvier>>,

crates/sui-bridge/src/e2e_tests/test_utils.rs

@@ -804,6 +804,7 @@ pub(crate) async fn start_bridge_cluster(
             },
             metrics_key_pair: default_ed25519_key_pair(),
             metrics: None,
+            watchdog_config: None,
         };
         // Spawn bridge node in memory
         handles.push(

crates/sui-bridge/src/eth_client.rs

@@ -36,6 +36,10 @@ impl EthClient<MeteredEthHttpProvier> {
         self_.describe().await?;
         Ok(self_)
     }
+
+    pub fn provider(&self) -> Arc<Provider<MeteredEthHttpProvier>> {
+        Arc::new(self.provider.clone())
+    }
 }
 
 #[cfg(test)]

crates/sui-bridge/src/lib.rs

@@ -19,6 +19,7 @@ pub mod node;
 pub mod orchestrator;
 pub mod server;
 pub mod storage;
+pub mod sui_bridge_watchdog;
 pub mod sui_client;
 pub mod sui_syncer;
 pub mod sui_transaction_builder;

crates/sui-bridge/src/node.rs

@@ -1,9 +1,20 @@
 // Copyright (c) Mysten Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
 
+use crate::config::WatchdogConfig;
 use crate::crypto::BridgeAuthorityPublicKeyBytes;
+use crate::metered_eth_provider::MeteredEthHttpProvier;
+use crate::sui_bridge_watchdog::eth_bridge_status::EthBridgeStatus;
+use crate::sui_bridge_watchdog::eth_vault_balance::EthVaultBalance;
+use crate::sui_bridge_watchdog::metrics::WatchdogMetrics;
+use crate::sui_bridge_watchdog::sui_bridge_status::SuiBridgeStatus;
+use crate::sui_bridge_watchdog::total_supplies::TotalSupplies;
+use crate::sui_bridge_watchdog::{BridgeWatchDog, Observable};
+use crate::sui_client::SuiBridgeClient;
 use crate::types::BridgeCommittee;
-use crate::utils::{get_committee_voting_power_by_name, get_validator_names_by_pub_keys};
+use crate::utils::{
+    get_committee_voting_power_by_name, get_eth_contract_addresses, get_validator_names_by_pub_keys,
+};
 use crate::{
     action_executor::BridgeActionExecutor,
     client::bridge_authority_aggregator::BridgeAuthorityAggregator,
@@ -18,6 +29,7 @@ use crate::{
     sui_syncer::SuiSyncer,
 };
 use arc_swap::ArcSwap;
+use ethers::providers::Provider;
 use ethers::types::Address as EthAddress;
 use mysten_metrics::spawn_logged_monitored_task;
 use std::collections::BTreeMap;
@@ -45,6 +57,7 @@ pub async fn run_bridge_node(
 ) -> anyhow::Result<JoinHandle<()>> {
     init_all_struct_tags();
     let metrics = Arc::new(BridgeMetrics::new(&prometheus_registry));
+    let watchdog_config = config.watchdog_config.clone();
     let (server_config, client_config) = config.validate(metrics.clone()).await?;
     let sui_chain_identifier = server_config
         .sui_client
@@ -73,6 +86,19 @@ pub async fn run_bridge_node(
             .await
             .expect("Failed to get committee"),
     );
+    let mut handles = vec![];
+
+    // Start watchdog
+    let eth_provider = server_config.eth_client.provider();
+    let eth_bridge_proxy_address = server_config.eth_bridge_proxy_address;
+    let sui_client = server_config.sui_client.clone();
+    handles.push(spawn_logged_monitored_task!(start_watchdog(
+        watchdog_config,
+        &prometheus_registry,
+        eth_provider,
+        eth_bridge_proxy_address,
+        sui_client
+    )));
 
     // Update voting right metrics
     // Before reconfiguration happens we only set it once when the node starts
@@ -84,19 +110,18 @@ pub async fn run_bridge_node(
         .await?;
 
     // Start Client
-    let _handles = if let Some(client_config) = client_config {
+    if let Some(client_config) = client_config {
         let committee_keys_to_names =
             Arc::new(get_validator_names_by_pub_keys(&committee, &sui_system).await);
-        start_client_components(
+        let client_components = start_client_components(
             client_config,
             committee.clone(),
             committee_keys_to_names,
             metrics.clone(),
         )
-        .await
-    } else {
-        Ok(vec![])
-    }?;
+        .await?;
+        handles.extend(client_components);
+    }
 
     let committee_name_mapping = get_committee_voting_power_by_name(&committee, &sui_system).await;
     for (name, voting_power) in committee_name_mapping.into_iter() {
@@ -125,6 +150,56 @@ pub async fn run_bridge_node(
     ))
 }
 
+async fn start_watchdog(
+    watchdog_config: Option<WatchdogConfig>,
+    registry: &prometheus::Registry,
+    eth_provider: Arc<Provider<MeteredEthHttpProvier>>,
+    eth_bridge_proxy_address: EthAddress,
+    sui_client: Arc<SuiBridgeClient>,
+) {
+    let watchdog_metrics = WatchdogMetrics::new(registry);
+    let (_committee_address, _limiter_address, vault_address, _config_address, weth_address) =
+        get_eth_contract_addresses(eth_bridge_proxy_address, &eth_provider)
+            .await
+            .unwrap_or_else(|e| panic!("get_eth_contract_addresses should not fail: {}", e));
+
+    let eth_vault_balance = EthVaultBalance::new(
+        eth_provider.clone(),
+        vault_address,
+        weth_address,
+        watchdog_metrics.eth_vault_balance.clone(),
+    );
+
+    let eth_bridge_status = EthBridgeStatus::new(
+        eth_provider,
+        eth_bridge_proxy_address,
+        watchdog_metrics.eth_bridge_paused.clone(),
+    );
+
+    let sui_bridge_status = SuiBridgeStatus::new(
+        sui_client.clone(),
+        watchdog_metrics.sui_bridge_paused.clone(),
+    );
+
+    let mut observables: Vec<Box<dyn Observable + Send + Sync>> = vec![
+        Box::new(eth_vault_balance),
+        Box::new(eth_bridge_status),
+        Box::new(sui_bridge_status),
+    ];
+    if let Some(watchdog_config) = watchdog_config {
+        if !watchdog_config.total_supplies.is_empty() {
+            let total_supplies = TotalSupplies::new(
+                Arc::new(sui_client.sui_client().clone()),
+                watchdog_config.total_supplies,
+                watchdog_metrics.total_supplies.clone(),
+            );
+            observables.push(Box::new(total_supplies));
+        }
+    }
+
+    BridgeWatchDog::new(observables).run().await
+}
+
 // TODO: is there a way to clean up the overrides after it's stored in DB?
 async fn start_client_components(
     client_config: BridgeClientConfig,
@@ -503,6 +578,7 @@ mod tests {
             db_path: None,
             metrics_key_pair: default_ed25519_key_pair(),
             metrics: None,
+            watchdog_config: None,
         };
         // Spawn bridge node in memory
         let _handle = run_bridge_node(
@@ -569,6 +645,7 @@ mod tests {
             db_path: Some(db_path),
             metrics_key_pair: default_ed25519_key_pair(),
             metrics: None,
+            watchdog_config: None,
         };
         // Spawn bridge node in memory
         let _handle = run_bridge_node(
@@ -646,6 +723,7 @@ mod tests {
             db_path: Some(db_path),
             metrics_key_pair: default_ed25519_key_pair(),
             metrics: None,
+            watchdog_config: None,
         };
         // Spawn bridge node in memory
         let _handle = run_bridge_node(