feat(dashboard): wire monitoring and indexer tasks through strata-tasks

Author: krsnapaudel

backend/Cargo.lock

@@ -8,9 +8,11 @@
   path = "src/main.rs"
 
 [dependencies]
+  anyhow.workspace         = true
   status-bridge.workspace  = true
   status-config.workspace  = true
   status-network.workspace = true
+  strata-tasks.workspace   = true
 
   axum.workspace               = true
   tokio.workspace              = true

backend/bin/dashboard/Cargo.toml

@@ -1,59 +1,96 @@
+use std::{net::SocketAddr, sync::Arc, time::Duration};
+
+use anyhow::Result;
 use axum::{routing::get, Router};
-use std::{net::SocketAddr, sync::Arc};
-use tokio::net::TcpListener;
+use status_bridge::{
+    bridge_monitoring_task, get_bridge_status, run_withdrawal_indexer, BridgeMonitoringContext,
+};
+use status_config::Config;
+use status_network::{fetch_statuses_task, get_network_status, NetworkMonitoringContext};
+use strata_tasks::TaskManager;
+use tokio::{net::TcpListener, runtime};
 use tower_http::cors::{Any, CorsLayer};
 use tracing::info;
 
-use status_bridge::{bridge_monitoring_task, get_bridge_status, BridgeMonitoringContext};
-use status_config::Config;
-use status_network::{fetch_statuses_task, get_network_status, NetworkMonitoringContext};
+const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(5);
 
-#[tokio::main]
-async fn main() -> Result<(), Box<dyn std::error::Error>> {
+fn main() -> Result<()> {
     tracing_subscriber::fmt::fmt()
         .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
         .init();
 
-    // Load configuration from TOML
+    let runtime = runtime::Builder::new_multi_thread()
+        .enable_all()
+        .thread_name("status-dashboard-rt")
+        .build()?;
+
     let config_path = std::env::var("CONFIG_PATH").unwrap_or_else(|_| "config.toml".to_string());
     let config = Arc::new(Config::load_from_path(&config_path));
 
-    let cors = CorsLayer::new().allow_origin(Any);
+    let task_manager = TaskManager::new(runtime.handle().clone());
+    let executor = task_manager.create_executor();
 
     let network_context = Arc::new(NetworkMonitoringContext::new(config.network().clone()));
-    let network_context_clone = Arc::clone(&network_context);
-
-    tokio::spawn(async move {
-        fetch_statuses_task(network_context_clone).await;
-    });
-
-    // Bridge monitoring
     let bridge_context = Arc::new(BridgeMonitoringContext::new(config.bridge().clone()));
-    let bridge_context_clone = Arc::clone(&bridge_context);
-
-    tokio::spawn(async move {
-        bridge_monitoring_task(bridge_context_clone).await;
-    });
 
+    let cors = CorsLayer::new().allow_origin(Any);
     let app = Router::new()
         .route(
             "/api/status",
-            get(move || get_network_status(network_context)),
+            get({
+                let network_context = Arc::clone(&network_context);
+                move || get_network_status(Arc::clone(&network_context))
+            }),
         )
         .route(
             "/api/bridge_status",
-            get(move || get_bridge_status(bridge_context)),
+            get({
+                let bridge_context = Arc::clone(&bridge_context);
+                move || get_bridge_status(Arc::clone(&bridge_context))
+            }),
         )
         .layer(cors);
 
     let addr = SocketAddr::from((
         config.server().host().parse::<std::net::IpAddr>()?,
         config.server().port(),
     ));
-    info!(%addr, "Server running at http://{}", addr);
 
-    let listener = TcpListener::bind(addr).await?;
-    axum::serve(listener, app).await?;
+    executor.spawn_critical_async_with_shutdown("network-monitoring", {
+        let network_context = Arc::clone(&network_context);
+        move |shutdown| async move { fetch_statuses_task(network_context, shutdown).await }
+    });
+
+    executor.spawn_critical_async_with_shutdown("bridge-monitoring", {
+        let bridge_context = Arc::clone(&bridge_context);
+        move |shutdown| async move { bridge_monitoring_task(bridge_context, shutdown).await }
+    });
+
+    executor.spawn_critical_async_with_shutdown("withdrawal-indexer", {
+        let datadir = config.datadir().to_path_buf();
+        let cfg = config.withdrawal_indexer().clone();
+        move |shutdown| async move { run_withdrawal_indexer(&datadir, cfg, shutdown).await }
+    });
+
+    executor.spawn_critical_async_with_shutdown("http-server", {
+        move |shutdown| async move {
+            let listener = TcpListener::bind(addr).await?;
+            info!(%addr, "Server running at http://{}", addr);
+            let shutdown_future = async move {
+                shutdown.wait_for_shutdown().await;
+            };
+
+            axum::serve(listener, app)
+                .with_graceful_shutdown(shutdown_future)
+                .await?;
+
+            Ok(())
+        }
+    });
+
+    task_manager.start_signal_listeners();
+    task_manager.monitor(Some(SHUTDOWN_TIMEOUT))?;
 
+    info!("Exiting status dashboard backend");
     Ok(())
 }

backend/bin/dashboard/src/main.rs

@@ -1,3 +1,4 @@
+use anyhow::Result;
 use axum::Json;
 use bitcoin::{secp256k1::PublicKey, Txid};
 
@@ -10,6 +11,7 @@ use strata_bridge_rpc::types::{
 };
 
 use strata_primitives::buf::Buf32;
+use strata_tasks::ShutdownGuard;
 
 use super::{
     cache::BridgeStatusCache,
@@ -283,7 +285,10 @@ async fn determine_reimbursements_to_purge(
 }
 
 /// Periodically fetch bridge status and update bridge cache
-pub async fn bridge_monitoring_task(context: Arc<BridgeMonitoringContext>) {
+pub async fn bridge_monitoring_task(
+    context: Arc<BridgeMonitoringContext>,
+    shutdown: ShutdownGuard,
+) -> Result<()> {
     let mut interval = interval(Duration::from_secs(
         context.config.status_refetch_interval(),
     ));
@@ -292,6 +297,11 @@ pub async fn bridge_monitoring_task(context: Arc<BridgeMonitoringContext>) {
     let rpc_manager = RpcClientManager::new(&context.config);
 
     loop {
+        tokio::select! {
+            _ = shutdown.wait_for_shutdown() => break,
+            _ = interval.tick() => {}
+        }
+
         // Fetch all data without holding lock
 
         // Bridge operator status
@@ -418,10 +428,9 @@ pub async fn bridge_monitoring_task(context: Arc<BridgeMonitoringContext>) {
             context.status_available.store(true, Ordering::Release);
             context.initial_status_query_complete.notify_waiters();
         }
-
-        // Wait for next interval
-        interval.tick().await;
     }
+
+    Ok(())
 }
 
 /// Fetch operator status

backend/crates/bridge/src/status.rs

@@ -87,11 +87,12 @@ where
         "withdrawal indexer starting"
     );
     loop {
-        // Process the tick `Result<_, IndexerError>` inside the select arm
-        // rather than binding it. `IndexerError` transitively wraps
-        // `Box<dyn Error>` (no `+ Send + Sync`) via `typed_sled::CodecError`,
-        // so it must not survive into the next await — only the `Send`
-        // `more_work` flag escapes.
+        // The indexer can stop mid-tick because writes are atomic per event
+        // and replay is idempotent on restart. Process the tick result
+        // inside the select arm rather than binding it: `IndexerError`
+        // transitively wraps `Box<dyn Error>` (no `+ Send + Sync`) via
+        // `typed_sled::CodecError`, so it must not survive into the next
+        // await — only the `Send` `more_work` flag escapes.
         let mut more_work = false;
         tokio::select! {
             _ = shutdown.wait_for_shutdown() => break,

backend/crates/bridge/src/withdrawal_indexer/task.rs

@@ -7,8 +7,10 @@
   path = "src/lib.rs"
 
 [dependencies]
+  anyhow.workspace        = true
   status-config.workspace = true
   status-utils.workspace  = true
+  strata-tasks.workspace  = true
 
   axum.workspace       = true
   jsonrpsee.workspace  = true

backend/crates/network/Cargo.toml

@@ -1,7 +1,9 @@
+use anyhow::Result;
 use axum::Json;
 use jsonrpsee::core::client::ClientT;
 use jsonrpsee::http_client::HttpClient;
 use std::sync::Arc;
+use strata_tasks::ShutdownGuard;
 use tokio::time::{interval, sleep, Duration};
 use tracing::{error, info};
 
@@ -58,7 +60,10 @@ async fn check_bundler_health(
 }
 
 /// Periodically fetches real statuses
-pub async fn fetch_statuses_task(context: Arc<NetworkMonitoringContext>) {
+pub async fn fetch_statuses_task(
+    context: Arc<NetworkMonitoringContext>,
+    shutdown: ShutdownGuard,
+) -> Result<()> {
     info!("Fetching statuses...");
     let mut interval = interval(Duration::from_secs(
         context.config.status_refetch_interval(),
@@ -68,7 +73,10 @@ pub async fn fetch_statuses_task(context: Arc<NetworkMonitoringContext>) {
     let http_client = reqwest::Client::new();
 
     loop {
-        interval.tick().await;
+        tokio::select! {
+            _ = shutdown.wait_for_shutdown() => break,
+            _ = interval.tick() => {}
+        }
 
         let sequencer =
             call_rpc_status(&sequencer_client, context.config.sequencer_retry_policy()).await;
@@ -96,6 +104,8 @@ pub async fn fetch_statuses_task(context: Arc<NetworkMonitoringContext>) {
             context.initial_status_query_complete.notify_waiters();
         }
     }
+
+    Ok(())
 }
 
 /// Handler to get the current network status