feat: add error logs for model insert error

Author: gusinacio

common/src/indexer_service/http/indexer_service.rs

@@ -36,6 +36,7 @@ use tracing::{info, info_span};
 
 use crate::escrow_accounts::EscrowAccounts;
 use crate::escrow_accounts::EscrowAccountsError;
+use crate::tap::{ValueCheckReceiver, ValueCheckSender};
 use crate::{
     address::public_key,
     indexer_service::http::static_subgraph::static_subgraph_request_handler,
@@ -176,6 +177,8 @@ where
     pub release: IndexerServiceRelease,
     pub url_namespace: &'static str,
     pub extra_routes: Router<Arc<IndexerServiceState<I>>>,
+    pub value_check_receiver: ValueCheckReceiver,
+    pub value_check_sender: ValueCheckSender,
 }
 
 pub struct IndexerServiceState<I>
@@ -190,6 +193,7 @@ where
     // tap
     pub escrow_accounts: Eventual<EscrowAccounts>,
     pub domain_separator: Eip712Domain,
+    pub value_check_sender: ValueCheckSender,
 }
 
 pub struct IndexerService {}
@@ -311,6 +315,7 @@ impl IndexerService {
             domain_separator.clone(),
             timestamp_error_tolerance,
             receipt_max_value,
+            options.value_check_receiver,
         )
         .await;
 
@@ -327,6 +332,7 @@ impl IndexerService {
             service_impl: Arc::new(options.service_impl),
             escrow_accounts,
             domain_separator,
+            value_check_sender: options.value_check_sender,
         });
 
         // Rate limits by allowing bursts of 10 requests and requiring 100ms of

common/src/tap.rs

@@ -6,6 +6,7 @@ use crate::tap::checks::deny_list_check::DenyListCheck;
 use crate::tap::checks::receipt_max_val_check::ReceiptMaxValueCheck;
 use crate::tap::checks::sender_balance_check::SenderBalanceCheck;
 use crate::tap::checks::timestamp_check::TimestampCheck;
+use crate::tap::checks::value_check::MinimumValue;
 use crate::{escrow_accounts::EscrowAccounts, prelude::Allocation};
 use alloy::dyn_abi::Eip712Domain;
 use alloy::primitives::Address;
@@ -24,6 +25,11 @@ use tracing::error;
 mod checks;
 mod receipt_store;
 
+pub use checks::value_check::{
+    create_value_check, CostModelSource, ValueCheckReceiver, ValueCheckSender,
+};
+
+#[derive(Clone)]
 pub struct IndexerTapContext {
     domain_separator: Arc<Eip712Domain>,
     receipt_producer: Sender<DatabaseReceipt>,
@@ -44,6 +50,7 @@ impl IndexerTapContext {
         domain_separator: Eip712Domain,
         timestamp_error_tolerance: Duration,
         receipt_max_value: u128,
+        value_check_receiver: ValueCheckReceiver,
     ) -> Vec<ReceiptCheck> {
         vec![
             Arc::new(AllocationEligible::new(indexer_allocations)),
@@ -54,6 +61,7 @@ impl IndexerTapContext {
             Arc::new(TimestampCheck::new(timestamp_error_tolerance)),
             Arc::new(DenyListCheck::new(pgpool, escrow_accounts, domain_separator).await),
             Arc::new(ReceiptMaxValueCheck::new(receipt_max_value)),
+            Arc::new(MinimumValue::new(value_check_receiver)),
         ]
     }
 

common/src/tap/checks/value_check.rs

@@ -12,7 +12,10 @@ use std::{
     time::Duration,
 };
 use thegraph_core::DeploymentId;
-use tokio::{sync::mpsc::Receiver, task::JoinHandle};
+use tokio::{
+    sync::mpsc::{Receiver, Sender},
+    task::JoinHandle,
+};
 use ttl_cache::TtlCache;
 
 use tap_core::{
@@ -31,10 +34,39 @@ pub struct MinimumValue {
     query_handle: JoinHandle<()>,
 }
 
+#[derive(Clone)]
+pub struct ValueCheckSender {
+    pub tx_cost_model: Sender<CostModelSource>,
+    pub tx_query: Sender<AgoraQuery>,
+}
+
+pub struct ValueCheckReceiver {
+    rx_cost_model: Receiver<CostModelSource>,
+    rx_query: Receiver<AgoraQuery>,
+}
+
+pub fn create_value_check(size: usize) -> (ValueCheckSender, ValueCheckReceiver) {
+    let (tx_cost_model, rx_cost_model) = tokio::sync::mpsc::channel(size);
+    let (tx_query, rx_query) = tokio::sync::mpsc::channel(size);
+
+    (
+        ValueCheckSender {
+            tx_query,
+            tx_cost_model,
+        },
+        ValueCheckReceiver {
+            rx_cost_model,
+            rx_query,
+        },
+    )
+}
+
 impl MinimumValue {
     pub fn new(
-        mut rx_cost_model: Receiver<CostModelSource>,
-        mut rx_query: Receiver<AgoraQuery>,
+        ValueCheckReceiver {
+            mut rx_query,
+            mut rx_cost_model,
+        }: ValueCheckReceiver,
     ) -> Self {
         let cost_model_cache = Arc::new(Mutex::new(HashMap::<DeploymentId, CostModelCache>::new()));
         let query_ids = Arc::new(Mutex::new(HashMap::new()));
@@ -48,7 +80,12 @@ impl MinimumValue {
                         let deployment_id = value.deployment_id;
 
                         if let Some(query) = cache.lock().unwrap().get_mut(&deployment_id) {
-                            let _ = query.insert_model(value);
+                            let _ = query.insert_model(value).inspect_err(|err| {
+                                tracing::error!(
+                                    "Error while compiling cost model for deployment id {}. Error: {}",
+                                    deployment_id, err
+                                )
+                            });
                         } else {
                             match CostModelCache::new(value) {
                                 Ok(value) => {
@@ -172,9 +209,9 @@ pub struct AgoraQuery {
 
 #[derive(Clone, Eq, Hash, PartialEq)]
 pub struct CostModelSource {
-    deployment_id: DeploymentId,
-    model: String,
-    variables: String,
+    pub deployment_id: DeploymentId,
+    pub model: String,
+    pub variables: String,
 }
 
 pub struct CostModelCache {

service/src/routes/cost.rs

@@ -12,6 +12,7 @@ use prometheus::{
     register_counter, register_counter_vec, register_histogram, register_histogram_vec, Counter,
     CounterVec, Histogram, HistogramVec,
 };
+use indexer_common::tap::CostModelSource;
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
 use thegraph_core::DeploymentId;
@@ -66,6 +67,16 @@ pub struct GraphQlCostModel {
     pub variables: Option<Value>,
 }
 
+impl From<CostModel> for CostModelSource {
+    fn from(value: CostModel) -> Self {
+        Self {
+            deployment_id: value.deployment,
+            model: value.model.unwrap_or_default(),
+            variables: value.variables.unwrap_or_default().to_string(),
+        }
+    }
+}
+
 impl From<CostModel> for GraphQlCostModel {
     fn from(model: CostModel) -> Self {
         Self {
@@ -127,8 +138,20 @@ impl Query {
         ctx: &Context<'_>,
         deployment_ids: Vec<DeploymentId>,
     ) -> Result<Vec<GraphQlCostModel>, anyhow::Error> {
-        let pool = &ctx.data_unchecked::<Arc<SubgraphServiceState>>().database;
+        let state = &ctx.data_unchecked::<Arc<SubgraphServiceState>>();
+
+        let cost_model_sender = &state.value_check_sender;
+
+        let pool = &state.database;
         let cost_models = database::cost_models(pool, &deployment_ids).await?;
+
+        for model in &cost_models {
+            let _ = cost_model_sender
+                .tx_cost_model
+                .send(CostModelSource::from(model.clone()))
+                .await;
+        }
+
         Ok(cost_models.into_iter().map(|m| m.into()).collect())
     }
 
@@ -137,10 +160,20 @@ impl Query {
         ctx: &Context<'_>,
         deployment_id: DeploymentId,
     ) -> Result<Option<GraphQlCostModel>, anyhow::Error> {
+
+        let state = &ctx.data_unchecked::<Arc<SubgraphServiceState>>();
+        let cost_model_sender = &state.value_check_sender;
         let pool = &ctx.data_unchecked::<Arc<SubgraphServiceState>>().database;
-        database::cost_model(pool, &deployment_id)
-            .await
-            .map(|model_opt| model_opt.map(GraphQlCostModel::from))
+        let model = database::cost_model(pool, &deployment_id).await?;
+
+        if let Some(model) = &model {
+            let _ = cost_model_sender
+                .tx_cost_model
+                .send(CostModelSource::from(model.clone()))
+                .await;
+        }
+
+        Ok(model.map(GraphQlCostModel::from))
     }
 }
 

service/src/service.rs

@@ -7,8 +7,9 @@ use std::time::Duration;
 use super::{config::Config, error::SubgraphServiceError, routes};
 use anyhow::anyhow;
 use axum::{async_trait, routing::post, Json, Router};
-use indexer_common::indexer_service::http::{
-    AttestationOutput, IndexerServiceImpl, IndexerServiceResponse,
+use indexer_common::{
+    indexer_service::http::{AttestationOutput, IndexerServiceImpl, IndexerServiceResponse},
+    tap::{create_value_check, ValueCheckSender},
 };
 use indexer_config::Config as MainConfig;
 use reqwest::Url;
@@ -67,6 +68,7 @@ pub struct SubgraphServiceState {
     pub graph_node_client: reqwest::Client,
     pub graph_node_status_url: String,
     pub graph_node_query_base_url: String,
+    pub value_check_sender: ValueCheckSender,
 }
 
 struct SubgraphService {
@@ -146,6 +148,9 @@ pub async fn run() -> anyhow::Result<()> {
     build_info::build_info!(fn build_info);
     let release = IndexerServiceRelease::from(build_info());
 
+    // arbitrary value
+    let (value_check_sender, value_check_receiver) = create_value_check(10);
+
     // Some of the subgraph service configuration goes into the so-called
     // "state", which will be passed to any request handler, middleware etc.
     // that is involved in serving requests
@@ -172,6 +177,7 @@ pub async fn run() -> anyhow::Result<()> {
             .expect("config must have `common.graph_node.query_url` set")
             .query_base_url
             .clone(),
+        value_check_sender: value_check_sender.clone(),
     });
 
     IndexerService::run(IndexerServiceOptions {
@@ -183,6 +189,8 @@ pub async fn run() -> anyhow::Result<()> {
             .route("/cost", post(routes::cost::cost))
             .route("/status", post(routes::status))
             .with_state(state),
+        value_check_receiver,
+        value_check_sender,
     })
     .await
 }