MTG-1110 Add request time out for DB query

README.md

@@ -60,5 +60,21 @@ The project's structure is a reflection of the following clean architecture prin
 
 The API specification is compatible with the standard DAS specification here https://github.com/metaplex-foundation/api-specifications
 
+
 ### Developing and running
+
+#### PR/Code requirements 
+1) CI/CD has code formating checker so use fmt before code commit: `cargo fmt`
+
+#### Run Integration tests
+Integration tests require Postgres db url, solana devnet and solana mainnet rpc urls.
+
+How to run with CLI:
+```shell
+DATABASE_TEST_URL='postgres://solana:solana@localhost:5432/aura_db' DEVNET_RPC_URL='https://api.devnet.solana.com' MAINNET_RPC_URL='https://api.mainnet-beta.solana.com' cargo test --features integration_tests
+
+
+
 Full documentation and contribution guidelines coming soon…
+
+

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_secs**
+

nft_ingester/src/api/api_impl.rs

@@ -4,6 +4,7 @@ use interface::json::{JsonDownloader, JsonPersister};
 use interface::processing_possibility::ProcessingPossibilityChecker;
 use interface::proofs::ProofChecker;
 use postgre_client::PgClient;
+use std::time::Duration;
 use std::{sync::Arc, time::Instant};
 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::InternalDbError(e.to_string()))?;
 
         self.metrics
             .set_latency(label, latency_timer.elapsed().as_millis() as f64);
@@ -714,30 +719,37 @@ where
         Self::validate_basic_pagination(&pagination, self.max_page_limit)?;
         Self::validate_options(&options, &query)?;
 
-        let res = search_assets(
-            pg_client,
-            rocks_db,
-            query,
-            sort_by,
-            pagination.limit.map_or(DEFAULT_LIMIT as u64, |l| l as u64),
-            pagination.page.map(|p| p as u64),
-            pagination.before,
-            pagination.after,
-            pagination.cursor,
-            options,
-            self.json_downloader.clone(),
-            self.json_persister.clone(),
-            self.json_middleware_config.max_urls_to_parse,
-            tasks,
-            self.account_balance_getter.clone(),
-            self.storage_service_base_path.clone(),
-            self.token_price_fetcher.clone(),
-            self.metrics.clone(),
-            &self.tree_gaps_checker,
-            &self.native_mint_pubkey,
+        let res = timeout(
+            Duration::from_secs(self.max_query_statement_timeout_sec),
+            search_assets(
+                pg_client,
+                rocks_db,
+                query,
+                sort_by,
+                pagination.limit.map_or(DEFAULT_LIMIT as u64, |l| l as u64),
+                pagination.page.map(|p| p as u64),
+                pagination.before,
+                pagination.after,
+                pagination.cursor,
+                options,
+                self.json_downloader.clone(),
+                self.json_persister.clone(),
+                self.json_middleware_config.max_urls_to_parse,
+                tasks,
+                self.account_balance_getter.clone(),
+                self.storage_service_base_path.clone(),
+                self.token_price_fetcher.clone(),
+                self.metrics.clone(),
+                &self.tree_gaps_checker,
+                &self.native_mint_pubkey,
+            ),
         )
-        .await?;
+        .await;
 
-        Ok(res)
+        match res {
+            Ok(Ok(res)) => Ok(res),
+            Ok(Err(e)) => Err(DasApiError::InternalDbError(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,
+    InternalDbError(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_secs: 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_secs,
     );
 
     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_secs,
         )
         .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_query_max_statement_timeout_secs,
             )
             .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,14 @@ 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_query_max_statement_timeout_secs: u64,
+
     #[clap(
         long,
         env = "INGESTER_METRICS_PORT",
@@ -430,6 +439,13 @@ 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_secs: u64,
+
     #[clap(
         long,
         env = "API_METRICS_PORT",