progress saver

MystenLabs · Sep 6, 2024 · a43cca4 · a43cca4
1 parent 375c654
commit a43cca4
Show file tree

Hide file tree

Showing 8 changed files with 416 additions and 94 deletions.
diff --git a/crates/sui-bridge-indexer/src/eth_bridge_indexer.rs b/crates/sui-bridge-indexer/src/eth_bridge_indexer.rs
@@ -154,10 +154,6 @@ impl Datasource<RawEthData> for EthSubscriptionDatasource {
     fn get_tasks_processed_checkpoints_metric(&self) -> &IntCounterVec {
         &self.indexer_metrics.tasks_processed_checkpoints
     }
-
-    fn get_live_task_checkpoint_metric(&self) -> &IntGaugeVec {
-        &self.indexer_metrics.live_task_current_checkpoint
-    }
 }
 
 pub struct EthSyncDatasource {
@@ -288,10 +284,6 @@ impl Datasource<RawEthData> for EthSyncDatasource {
     fn get_tasks_processed_checkpoints_metric(&self) -> &IntCounterVec {
         &self.indexer_metrics.tasks_processed_checkpoints
     }
-
-    fn get_live_task_checkpoint_metric(&self) -> &IntGaugeVec {
-        &self.indexer_metrics.live_task_current_checkpoint
-    }
 }
 
 #[derive(Clone)]

diff --git a/crates/sui-bridge-indexer/src/main.rs b/crates/sui-bridge-indexer/src/main.rs
@@ -24,7 +24,10 @@ use sui_bridge_indexer::config::IndexerConfig;
 use sui_bridge_indexer::eth_bridge_indexer::EthDataMapper;
 use sui_bridge_indexer::metrics::BridgeIndexerMetrics;
 use sui_bridge_indexer::postgres_manager::{get_connection_pool, read_sui_progress_store};
-use sui_bridge_indexer::storage::PgBridgePersistent;
+use sui_bridge_indexer::storage::{
+    OutOfOrderSaveAfterDurationPolicy, PgBridgePersistent, ProgressSavingPolicy,
+    SaveAfterDurationPolicy,
+};
 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;
@@ -72,7 +75,20 @@ async fn main() -> Result<()> {
     let bridge_metrics = Arc::new(BridgeMetrics::new(&registry));
 
     let db_url = config.db_url.clone();
-    let datastore = PgBridgePersistent::new(get_connection_pool(db_url.clone()).await);
+    let datastore = PgBridgePersistent::new(
+        get_connection_pool(db_url.clone()).await,
+        ProgressSavingPolicy::SaveAfterDuration(SaveAfterDurationPolicy::new(
+            tokio::time::Duration::from_secs(30),
+        )),
+        indexer_meterics.clone(),
+    );
+    let datastore_with_out_of_order_source = PgBridgePersistent::new(
+        get_connection_pool(db_url.clone()).await,
+        ProgressSavingPolicy::OutOfOrderSaveAfterDuration(OutOfOrderSaveAfterDurationPolicy::new(
+            tokio::time::Duration::from_secs(30),
+        )),
+        indexer_meterics.clone(),
+    );
 
     let eth_client: Arc<EthClient<MeteredEthHttpProvier>> = Arc::new(
         EthClient::<MeteredEthHttpProvier>::new(
@@ -119,7 +135,7 @@ async fn main() -> Result<()> {
         EthDataMapper {
             metrics: indexer_meterics.clone(),
         },
-        datastore.clone(),
+        datastore,
     )
     .with_backfill_strategy(BackfillStrategy::Partitioned { task_size: 1000 })
     .disable_live_task()
@@ -146,7 +162,7 @@ async fn main() -> Result<()> {
         SuiBridgeDataMapper {
             metrics: indexer_meterics.clone(),
         },
-        datastore,
+        datastore_with_out_of_order_source,
     )
     .build();
     indexer.start().await?;

diff --git a/crates/sui-bridge-indexer/src/metrics.rs b/crates/sui-bridge-indexer/src/metrics.rs
@@ -23,7 +23,7 @@ pub struct BridgeIndexerMetrics {
     pub(crate) last_synced_eth_block: IntGauge,
     pub(crate) tasks_remaining_checkpoints: IntGaugeVec,
     pub(crate) tasks_processed_checkpoints: IntCounterVec,
-    pub(crate) live_task_current_checkpoint: IntGaugeVec,
+    pub(crate) tasks_current_checkpoints: IntGaugeVec,
 }
 
 impl BridgeIndexerMetrics {
@@ -115,9 +115,9 @@ impl BridgeIndexerMetrics {
                 registry,
             )
             .unwrap(),
-            live_task_current_checkpoint: register_int_gauge_vec_with_registry!(
-                "bridge_indexer_live_task_current_checkpoint",
-                "Current checkpoint of live task",
+            tasks_current_checkpoints: register_int_gauge_vec_with_registry!(
+                "bridge_indexer_tasks_current_checkpoints",
+                "Current checkpoint for each task",
                 &["task_name"],
                 registry,
             )