wip(refactor(bridge-indexer)): CancellationToken -> Service

Author: amnn

Cargo.lock

@@ -8,7 +8,6 @@ edition = "2024"
 
 [dependencies]
 tokio.workspace = true
-tokio-util.workspace = true
 anyhow.workspace = true
 sui-bridge.workspace = true
 sui-bridge-schema.workspace = true

crates/sui-bridge-indexer-alt/Cargo.toml

@@ -12,9 +12,9 @@ use sui_bridge_indexer_alt::metrics::BridgeIndexerMetrics;
 use sui_bridge_schema::MIGRATIONS;
 use sui_indexer_alt_framework::ingestion::{ClientArgs, ingestion_client::IngestionClientArgs};
 use sui_indexer_alt_framework::postgres::DbArgs;
+use sui_indexer_alt_framework::service::Error;
 use sui_indexer_alt_framework::{Indexer, IndexerArgs};
 use sui_indexer_alt_metrics::{MetricsArgs, MetricsService};
-use tokio_util::sync::CancellationToken;
 use url::Url;
 
 #[derive(Parser)]
@@ -49,18 +49,14 @@ async fn main() -> Result<(), anyhow::Error> {
         remote_store_url,
     } = Args::parse();
 
-    let cancel = CancellationToken::new();
+    let is_bounded_job = indexer_args.last_checkpoint.is_some();
     let registry = Registry::new_custom(Some("bridge".into()), None)
         .context("Failed to create Prometheus registry.")?;
 
     // Initialize bridge-specific metrics
     let bridge_metrics = BridgeIndexerMetrics::new(&registry);
 
-    let metrics = MetricsService::new(
-        MetricsArgs { metrics_address },
-        registry,
-        cancel.child_token(),
-    );
+    let metrics = MetricsService::new(MetricsArgs { metrics_address }, registry);
 
     let metrics_prefix = None;
     let mut indexer = Indexer::new_from_pg(
@@ -78,7 +74,6 @@ async fn main() -> Result<(), anyhow::Error> {
         Some(&MIGRATIONS),
         metrics_prefix,
         metrics.registry(),
-        cancel.clone(),
     )
     .await?;
 
@@ -104,11 +99,23 @@ async fn main() -> Result<(), anyhow::Error> {
         .concurrent_pipeline(ErrorTransactionHandler, Default::default())
         .await?;
 
-    let h_indexer = indexer.run().await?;
-    let h_metrics = metrics.run().await?;
+    let s_indexer = indexer.run().await?;
+    let s_metrics = metrics.run().await?;
 
-    let _ = h_indexer.await;
-    cancel.cancel();
-    let _ = h_metrics.await;
-    Ok(())
+    match s_indexer.attach(s_metrics).main().await {
+        Ok(()) => Ok(()),
+        Err(Error::Terminated) => {
+            if is_bounded_job {
+                std::process::exit(1);
+            } else {
+                Ok(())
+            }
+        }
+        Err(Error::Aborted) => {
+            std::process::exit(1);
+        }
+        Err(Error::Task(_)) => {
+            std::process::exit(2);
+        }
+    }
 }

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

@@ -15,7 +15,6 @@ path = "src/main.rs"
 
 [dependencies]
 sui-config.workspace = true
-sui-futures.workspace = true
 sui-indexer-alt-framework.workspace = true
 sui-indexer-alt-framework-store-traits.workspace = true
 sui-indexer-alt-object-store.workspace = true

crates/sui-checkpoint-blob-indexer/Cargo.toml

@@ -8,11 +8,10 @@ use object_store::{
     gcp::GoogleCloudStorageBuilder, http::HttpBuilder, local::LocalFileSystem,
 };
 use sui_checkpoint_blob_indexer::{CheckpointBlobPipeline, EpochsPipeline};
-use sui_futures::service::Error;
+use sui_indexer_alt_framework::service::Error;
 use sui_indexer_alt_framework::{Indexer, IndexerArgs, ingestion::ClientArgs};
 use sui_indexer_alt_metrics::MetricsArgs;
 use sui_indexer_alt_object_store::ObjectStore;
-use tracing::error;
 use url::Url;
 
 #[derive(Debug, clap::Parser)]