MTG-1110 Add request time out for DB query

integration_tests/src/common.rs

@@ -179,6 +179,7 @@ impl TestSetup {
                 None,
             )),
             "11111111111111111111111111111111".to_string(),
+            120
         );
 
         let message_parser = MessageParser::new();

nft_ingester/readme.md

@@ -0,0 +1,16 @@
+
+
+
+
+
+
+
+## Tips
+
+To set a global limit on request execution for PG DB, you can use the URL parameter  **statement_timeout**.
+
+Example:
+`postgres://user:password@localhost/dbname?statement_timeout=2000`
+
+To limit only API requests use config option **api_query_max_statement_timeout_sec**
+

nft_ingester/src/api/api_impl.rs

@@ -5,6 +5,7 @@ use interface::processing_possibility::ProcessingPossibilityChecker;
 use interface::proofs::ProofChecker;
 use postgre_client::PgClient;
 use std::{sync::Arc, time::Instant};
+use std::time::Duration;
 use tokio::sync::Mutex;
 use tokio::task::{JoinError, JoinSet};
 
@@ -32,6 +33,7 @@ use metrics_utils::ApiMetricsConfig;
 use rocks_db::Storage;
 use serde_json::{json, Value};
 use solana_sdk::pubkey::Pubkey;
+use tokio::time::timeout;
 use usecase::validation::{validate_opt_pubkey, validate_pubkey};
 
 const MAX_ITEMS_IN_BATCH_REQ: usize = 1000;
@@ -60,6 +62,7 @@ where
     storage_service_base_path: Option<String>,
     token_price_fetcher: Arc<TPF>,
     native_mint_pubkey: String,
+    max_query_statement_timeout_sec: u64,
 }
 
 pub fn not_found() -> DasApiError {
@@ -90,6 +93,7 @@ where
         storage_service_base_path: Option<String>,
         token_price_fetcher: Arc<TPF>,
         native_mint_pubkey: String,
+        max_query_statement_timeout_sec: u64,
     ) -> Self {
         DasApi {
             pg_client,
@@ -105,6 +109,7 @@ where
             storage_service_base_path,
             token_price_fetcher,
             native_mint_pubkey,
+            max_query_statement_timeout_sec,
         }
     }
 
@@ -116,7 +121,7 @@ where
         self.pg_client
             .check_health()
             .await
-            .map_err(|_| DasApiError::InternalDdError)?;
+            .map_err(|e| DasApiError::InternalDdError(e.to_string()))?;
 
         self.metrics
             .set_latency(label, latency_timer.elapsed().as_millis() as f64);
@@ -714,7 +719,7 @@ where
         Self::validate_basic_pagination(&pagination, self.max_page_limit)?;
         Self::validate_options(&options, &query)?;
 
-        let res = search_assets(
+        let res = timeout(Duration::from_secs(self.max_query_statement_timeout_sec), search_assets(
             pg_client,
             rocks_db,
             query,
@@ -735,9 +740,16 @@ where
             self.metrics.clone(),
             &self.tree_gaps_checker,
             &self.native_mint_pubkey,
-        )
-        .await?;
-
-        Ok(res)
+        )).await;
+
+        match res {
+            Ok(Ok(res)) => Ok(res),
+            Ok(Err(e)) => {
+                Err(DasApiError::InternalDdError(e.to_string()))
+            },
+            Err(_) => {
+                Err(DasApiError::QueryTimedOut)
+            }
+        }
     }
 }

nft_ingester/src/api/error.rs

@@ -39,11 +39,13 @@ pub enum DasApiError {
     #[error("Page number is too big. Up to {0} pages are supported with this kind of pagination. Please use a different pagination(before/after/cursor).")]
     PageTooBig(usize),
     #[error("Internal DB error")]
-    InternalDdError,
+    InternalDdError(String),
     #[error("CannotServiceRequest")]
     CannotServiceRequest,
     #[error("MissingOwnerAddress")]
     MissingOwnerAddress,
+    #[error("QueryTimedOut")]
+    QueryTimedOut,
 }
 
 impl From<DasApiError> for jsonrpc_core::Error {

nft_ingester/src/api/service.rs

@@ -71,6 +71,7 @@ pub async fn start_api(
     account_balance_getter: Arc<AccountBalanceGetterImpl>,
     storage_service_base_url: Option<String>,
     native_mint_pubkey: String,
+    api_query_max_statement_timeout_sec: u64,
 ) -> Result<(), DasApiError> {
     let response_middleware = RpcResponseMiddleware {};
     let request_middleware = RpcRequestMiddleware::new(archives_dir);
@@ -127,6 +128,7 @@ pub async fn start_api(
             red_metrics,
         )),
         native_mint_pubkey,
+        api_query_max_statement_timeout_sec
     );
 
     run_api(

nft_ingester/src/bin/api/main.rs

@@ -155,6 +155,7 @@ pub async fn main() -> Result<(), IngesterError> {
             account_balance_getter,
             args.storage_service_base_url,
             args.native_mint_pubkey,
+            args.api_query_max_statement_timeout_sec
         )
         .await
         {

nft_ingester/src/bin/ingester/main.rs

@@ -306,6 +306,7 @@ pub async fn main() -> Result<(), IngesterError> {
                 account_balance_getter,
                 args.storage_service_base_url,
                 args.native_mint_pubkey,
+                args.api_max_query_statement_timeout_sec
             )
             .await
             {

nft_ingester/src/config.rs

@@ -110,6 +110,7 @@ pub struct IngesterClapArgs {
         requires = "rocks_backup_archives_dir"
     )]
     pub is_restore_rocks_db: bool,
+
     #[clap(long, env, help = "Rocks backup url")]
     pub rocks_backup_url: Option<String>,
     #[clap(long, env, help = "Rocks backup archives dir")]
@@ -251,6 +252,9 @@ pub struct IngesterClapArgs {
     #[clap(long, env, default_value = "500", help = "#grpc retry interval millis")]
     pub rpc_retry_interval_millis: u64,
 
+    #[clap(long, env, default_value = "120", help= "Specifies the maximum execution time of a SQL query for API.")]
+    pub api_max_query_statement_timeout_sec: u64,
+
     #[clap(
         long,
         env = "INGESTER_METRICS_PORT",
@@ -430,6 +434,9 @@ pub struct ApiClapArgs {
     #[clap(long, env, default_value = "/usr/src/app/heaps", help = "Heap path")]
     pub heap_path: String,
 
+    #[clap(long, default_value = "120", help= "Specifies the maximum execution time of a SQL query for API.")]
+    pub api_query_max_statement_timeout_sec: u64,
+
     #[clap(
         long,
         env = "API_METRICS_PORT",