chore: add monitor service back

Co-authored-by: Nagaprasadvr <[email protected]>

grpc-ingest/Cargo.toml

@@ -21,6 +21,7 @@ chrono = { workspace = true }
 clap = { workspace = true, features = ["cargo", "derive"] }
 thiserror = { workspace = true }
 das-core = { workspace = true }
+das-bubblegum = { workspace = true }
 digital_asset_types = { workspace = true }
 futures = { workspace = true }
 hex = { workspace = true }

grpc-ingest/src/config.rs

@@ -384,3 +384,35 @@ impl ConfigIngesterDownloadMetadata {
         1
     }
 }
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct ConfigMonitor {
+    pub postgres: ConfigPostgres,
+    pub rpc: String,
+    pub bubblegum: ConfigBubblegumVerify,
+}
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct ConfigBubblegumVerify {
+    #[serde(
+        default = "ConfigBubblegumVerify::default_report_interval",
+        deserialize_with = "deserialize_duration_str"
+    )]
+    pub _report_interval: Duration,
+    #[serde(default)]
+    pub only_trees: Option<Vec<String>>,
+    #[serde(
+        default = "ConfigBubblegumVerify::default_max_concurrency",
+        deserialize_with = "deserialize_usize_str"
+    )]
+    pub max_concurrency: usize,
+}
+
+impl ConfigBubblegumVerify {
+    pub const fn default_report_interval() -> Duration {
+        Duration::from_millis(5 * 60 * 1000)
+    }
+    pub const fn default_max_concurrency() -> usize {
+        20
+    }
+}

grpc-ingest/src/main.rs

@@ -6,12 +6,14 @@ use {
     },
     anyhow::Context,
     clap::{Parser, Subcommand},
+    config::ConfigMonitor,
     std::net::SocketAddr,
 };
 
 mod config;
 mod grpc;
 mod ingester;
+mod monitor;
 mod postgres;
 mod prom;
 mod redis;
@@ -42,6 +44,9 @@ enum ArgsAction {
     /// Run ingester process (process events from Redis)
     #[command(name = "ingester")]
     Ingester,
+    /// Run monitor process (verify bubblegum proofs)
+    #[command(name = "monitor")]
+    Monitor,
 }
 
 #[tokio::main]
@@ -58,7 +63,7 @@ async fn main() -> anyhow::Result<()> {
         prometheus_run_server(address)?;
     }
 
-    // Run grpc / ingester / download-metadata
+    // Run grpc / ingester / download-metadata / monitor
     match args.action {
         ArgsAction::Grpc => {
             let config = config_load::<ConfigGrpc>(&args.config)
@@ -72,5 +77,11 @@ async fn main() -> anyhow::Result<()> {
                 .with_context(|| format!("failed to parse config from: {}", args.config))?;
             ingester::run(config).await
         }
+        ArgsAction::Monitor => {
+            let config = config_load::<ConfigMonitor>(&args.config)
+                .await
+                .with_context(|| format!("failed to parse config from: {}", args.config))?;
+            monitor::run(config).await
+        }
     }
 }

grpc-ingest/src/monitor.rs

@@ -0,0 +1,50 @@
+use crate::postgres::create_pool;
+use crate::util::create_shutdown;
+use crate::{config::ConfigMonitor, prom::update_tree_proof_report};
+use das_bubblegum::{verify_bubblegum, BubblegumContext, VerifyArgs};
+use das_core::{Rpc, SolanaRpcArgs};
+use futures::stream::StreamExt;
+use tracing::{error, info};
+
+pub async fn run(config: ConfigMonitor) -> anyhow::Result<()> {
+    let mut shutdown = create_shutdown()?;
+    let database_pool = create_pool(config.postgres).await?;
+    let rpc = Rpc::from_config(&SolanaRpcArgs {
+        solana_rpc_url: config.rpc,
+    });
+
+    let bubblegum_verify = tokio::spawn(async move {
+        loop {
+            let bubblegum_context = BubblegumContext::new(database_pool.clone(), rpc.clone());
+            let verify_args = VerifyArgs {
+                only_trees: config.bubblegum.only_trees.clone(),
+                max_concurrency: config.bubblegum.max_concurrency,
+            };
+
+            match verify_bubblegum(bubblegum_context, verify_args).await {
+                Ok(mut reports_receiver) => {
+                    while let Some(report) = reports_receiver.recv().await {
+                        info!(
+                            report = ?report,
+                        );
+                        update_tree_proof_report(&report);
+                    }
+
+                    tokio::time::sleep(tokio::time::Duration::from_secs(600)).await;
+                }
+                Err(e) => {
+                    error!(
+                        message = "Error proof report recv",
+                        error = ?e
+                    );
+                }
+            }
+        }
+    });
+
+    if let Some(_signal) = shutdown.next().await {}
+
+    bubblegum_verify.abort();
+
+    Ok(())
+}

grpc-ingest/src/prom.rs

@@ -1,5 +1,6 @@
 use {
     crate::{redis::RedisStreamMessageError, version::VERSION as VERSION_INFO},
+    das_bubblegum::ProofReport,
     das_core::MetadataJsonTaskError,
     hyper::{
         server::conn::AddrStream,
@@ -391,3 +392,25 @@ pub fn program_transformer_task_status_inc(
         .with_label_values(&[stream, consumer, kind.to_str()])
         .inc()
 }
+
+pub fn update_tree_proof_report(report: &ProofReport) {
+    BUBBLEGUM_TREE_TOTAL_LEAVES
+        .with_label_values(&[&report.tree_pubkey.to_string()])
+        .set(report.total_leaves as i64);
+
+    BUBBLEGUM_TREE_INCORRECT_PROOFS
+        .with_label_values(&[&report.tree_pubkey.to_string()])
+        .set(report.incorrect_proofs as i64);
+
+    BUBBLEGUM_TREE_NOT_FOUND_PROOFS
+        .with_label_values(&[&report.tree_pubkey.to_string()])
+        .set(report.not_found_proofs as i64);
+
+    BUBBLEGUM_TREE_CORRECT_PROOFS
+        .with_label_values(&[&report.tree_pubkey.to_string()])
+        .set(report.correct_proofs as i64);
+
+    BUBBLEGUM_TREE_CORRUPT_PROOFS
+        .with_label_values(&[&report.tree_pubkey.to_string()])
+        .set(report.corrupt_proofs as i64);
+}