[iota-indexer]: merge rpc-tests branch into develop

.github/workflows/_rust_tests.yml

@@ -243,3 +243,6 @@ jobs:
           cargo nextest run --no-fail-fast --test-threads 8 --package iota-graphql-e2e-tests --features pg_integration
           cargo nextest run --no-fail-fast --test-threads 1 --package iota-cluster-test --test local_cluster_test --features pg_integration
           cargo nextest run --no-fail-fast --test-threads 1 --package iota-indexer --test ingestion_tests --features pg_integration
+          # Iota-indexer's RPC tests, which depend on a shared runtime, are incompatible with nextest due to its process-per-test execution model.
+          # cargo test, on the other hand, allows tests to share state and resources by default.
+          cargo test --profile simulator --package iota-indexer --test rpc-tests --features shared_test_runtime

crates/iota-cluster-test/src/cluster.rs

@@ -265,6 +265,7 @@ impl Cluster for LocalNewCluster {
                 fullnode_url.clone(),
                 ReaderWriterConfig::writer_mode(None),
                 data_ingestion_path.clone(),
+                None,
             )
             .await;
 
@@ -274,6 +275,7 @@ impl Cluster for LocalNewCluster {
                 fullnode_url.clone(),
                 ReaderWriterConfig::reader_mode(indexer_address.to_string()),
                 data_ingestion_path,
+                None,
             )
             .await;
         }

crates/iota-graphql-rpc/src/test_infra/cluster.rs

@@ -73,6 +73,7 @@ pub async fn start_cluster(
         true,
         Some(data_ingestion_path),
         cancellation_token.clone(),
+        None,
     )
     .await;
 
@@ -137,6 +138,7 @@ pub async fn serve_executor(
         true,
         Some(data_ingestion_path),
         cancellation_token.clone(),
+        Some(&graphql_connection_config.db_name()),
     )
     .await;
 

crates/iota-indexer/Cargo.toml

@@ -29,7 +29,6 @@ rayon.workspace = true
 regex.workspace = true
 secrecy = "0.8.0"
 serde.workspace = true
-serde_json.workspace = true
 serde_with.workspace = true
 tap.workspace = true
 tempfile.workspace = true
@@ -60,6 +59,7 @@ telemetry-subscribers.workspace = true
 
 [features]
 pg_integration = []
+shared_test_runtime = []
 default = ["postgres-feature"]
 postgres-feature = ["diesel/postgres", "diesel/postgres_backend"]
 mysql-feature = ["diesel/mysql", "diesel/mysql_backend", "dep:mysqlclient-sys"]
@@ -68,14 +68,17 @@ bundled-mysql = ["mysqlclient-sys?/bundled"]
 [dev-dependencies]
 # external dependencies
 rand.workspace = true
+serde_json.workspace = true
 
 # internal dependencies
 iota-config.workspace = true
 iota-genesis-builder.workspace = true
 iota-keys.workspace = true
 iota-move-build.workspace = true
 iota-swarm-config.workspace = true
+iota-test-transaction-builder.workspace = true
 simulacrum.workspace = true
+test-cluster.workspace = true
 
 [[bin]]
 name = "iota-indexer"

crates/iota-indexer/README.md

@@ -84,14 +84,41 @@ diesel database reset --database-url="postgres://postgres:postgrespw@localhost/i
 
 ### Running tests
 
-To run the tests, a running postgres instance is required. The crate provides following tests currently:
+To run the tests, a running postgres instance is required.
+
+```sh
+docker run --name iota-indexer-tests -e POSTGRES_PASSWORD=postgrespw -e POSTGRES_USER=postgres -e POSTGRES_DB=iota_indexer -d -p 5432:5432 postgres
+```
+
+The crate provides following tests currently:
 
 - unit tests for DB models (objects, events) which test the conversion between the database representation and the Rust representation of the objects and events.
 - unit tests for the DB query filters, which test the conversion of filters to the correct SQL queries.
 - integration tests (see [ingestion_tests](tests/ingestion_tests.rs)) to make sure the indexer correctly indexes transaction data from a full node by comparing the data in the database with the data received from the fullnode.
+- rpc tests (see [rpc-tests](tests/rpc-tests/main.rs))
+
+> [!NOTE]
+> rpc tests which relies on postgres for every test it applies migrations, we need to run tests sequencially to avoid errors
+
+```sh
+# run tests requiring only postgres integration
+cargo test --features pg_integration -- --test-threads 1
+# run rpc tests with shared runtime
+cargo test --profile simulator --features shared_test_runtime
+```
+
+For a better testing experience is possible to use [nextest](https://nexte.st/)
+
+> [!NOTE]
+> rpc tests which rely on a shared runtime are not supported with `nextest`
+>
+> This is because `cargo nextest` process-per-test execution model makes extremely difficult to share state and resources between tests.
+>
+> On the other hand `cargo test` does not run tests in separate processes by default. This means that tests can share state and resources.
 
 ```sh
-cargo nextest --features pg_integration --test-threads 1
+# run tests requiring only postgres integration
+cargo nextest run --features pg_integration --test-threads 1
 ```
 
 ## Steps to run locally with TiDB (experimental)

crates/iota-indexer/src/apis/governance_api.rs

@@ -2,10 +2,10 @@
 // Modifications Copyright (c) 2024 IOTA Stiftung
 // SPDX-License-Identifier: Apache-2.0
 
-use std::collections::BTreeMap;
+use std::{collections::BTreeMap, sync::Arc};
 
 use async_trait::async_trait;
-use cached::{SizedCache, proc_macro::cached};
+use cached::{Cached, SizedCache};
 use diesel::r2d2::R2D2Connection;
 use iota_json_rpc::{IotaRpcModule, governance_api::ValidatorExchangeRates};
 use iota_json_rpc_api::GovernanceReadApiServer;
@@ -23,6 +23,7 @@ use iota_types::{
     timelock::timelocked_staked_iota::TimelockedStakedIota,
 };
 use jsonrpsee::{RpcModule, core::RpcResult};
+use tokio::sync::Mutex;
 
 use crate::{errors::IndexerError, indexer_reader::IndexerReader};
 
@@ -32,19 +33,27 @@ const MAX_QUERY_STAKED_OBJECTS: usize = 1000;
 #[derive(Clone)]
 pub struct GovernanceReadApi<T: R2D2Connection + 'static> {
     inner: IndexerReader<T>,
+    exchange_rates_cache: Arc<Mutex<SizedCache<EpochId, Vec<ValidatorExchangeRates>>>>,
+    validators_apys_cache: Arc<Mutex<SizedCache<EpochId, BTreeMap<IotaAddress, f64>>>>,
 }
 
 impl<T: R2D2Connection + 'static> GovernanceReadApi<T> {
     pub fn new(inner: IndexerReader<T>) -> Self {
-        Self { inner }
+        Self {
+            inner,
+            exchange_rates_cache: Arc::new(Mutex::new(SizedCache::with_size(1))),
+            validators_apys_cache: Arc::new(Mutex::new(SizedCache::with_size(1))),
+        }
     }
 
     /// Get a validator's APY by its address
     pub async fn get_validator_apy(
         &self,
         address: &IotaAddress,
     ) -> Result<Option<f64>, IndexerError> {
-        let apys = validators_apys_map(self.get_validators_apy().await?);
+        let apys = self
+            .validators_apys_map(self.get_validators_apy().await?)
+            .await;
         Ok(apys.get(address).copied())
     }
 
@@ -261,6 +270,28 @@ impl<T: R2D2Connection + 'static> GovernanceReadApi<T> {
         }
         Ok(delegated_stakes)
     }
+
+    /// Cache a map representing the validators' APYs for this epoch
+    async fn validators_apys_map(&self, apys: ValidatorApys) -> BTreeMap<IotaAddress, f64> {
+        // check if the apys are already in the cache
+        if let Some(cached_apys) = self
+            .validators_apys_cache
+            .lock()
+            .await
+            .cache_get(&apys.epoch)
+        {
+            return cached_apys.clone();
+        }
+
+        let ret = BTreeMap::from_iter(apys.apys.iter().map(|x| (x.address, x.apy)));
+        // insert the apys into the cache
+        self.validators_apys_cache
+            .lock()
+            .await
+            .cache_set(apys.epoch, ret.clone());
+
+        ret
+    }
 }
 
 fn stake_status(
@@ -292,15 +323,31 @@ fn stake_status(
 /// Cached exchange rates for validators for the given epoch, the cache size is
 /// 1, it will be cleared when the epoch changes. rates are in descending order
 /// by epoch.
-#[cached(
-    type = "SizedCache<EpochId, Vec<ValidatorExchangeRates>>",
-    create = "{ SizedCache::with_size(1) }",
-    convert = "{ system_state_summary.epoch }",
-    result = true
-)]
 pub async fn exchange_rates(
     state: &GovernanceReadApi<impl R2D2Connection>,
     system_state_summary: &IotaSystemStateSummary,
+) -> Result<Vec<ValidatorExchangeRates>, IndexerError> {
+    let epoch = system_state_summary.epoch;
+
+    let mut cache = state.exchange_rates_cache.lock().await;
+
+    // Check if the exchange rates for the current epoch are cached
+    if let Some(cached_rates) = cache.cache_get(&epoch) {
+        return Ok(cached_rates.clone());
+    }
+
+    // Cache miss: compute exchange rates
+    let exchange_rates = compute_exchange_rates(state, system_state_summary).await?;
+
+    // Store in cache
+    cache.cache_set(epoch, exchange_rates.clone());
+
+    Ok(exchange_rates)
+}
+
+pub async fn compute_exchange_rates(
+    state: &GovernanceReadApi<impl R2D2Connection>,
+    system_state_summary: &IotaSystemStateSummary,
 ) -> Result<Vec<ValidatorExchangeRates>, IndexerError> {
     // Get validator rate tables
     let mut tables = vec![];
@@ -384,16 +431,6 @@ pub async fn exchange_rates(
     Ok(exchange_rates)
 }
 
-/// Cache a map representing the validators' APYs for this epoch
-#[cached(
-    type = "SizedCache<EpochId, BTreeMap<IotaAddress, f64>>",
-    create = "{ SizedCache::with_size(1) }",
-    convert = " {apys.epoch} "
-)]
-fn validators_apys_map(apys: ValidatorApys) -> BTreeMap<IotaAddress, f64> {
-    BTreeMap::from_iter(apys.apys.iter().map(|x| (x.address, x.apy)))
-}
-
 #[async_trait]
 impl<T: R2D2Connection + 'static> GovernanceReadApiServer for GovernanceReadApi<T> {
     async fn get_stakes_by_ids(

crates/iota-indexer/src/models/transactions.rs

@@ -268,18 +268,15 @@ impl StoredTransaction {
             None
         };
 
-        let effects = if options.show_effects {
-            let effects = self.try_into_iota_transaction_effects()?;
-            Some(effects)
-        } else {
-            None
-        };
+        let effects = options
+            .show_effects
+            .then(|| self.try_into_iota_transaction_effects())
+            .transpose()?;
 
-        let raw_transaction = if options.show_raw_input {
-            self.raw_transaction
-        } else {
-            Vec::new()
-        };
+        let raw_transaction = options
+            .show_raw_input
+            .then_some(self.raw_transaction)
+            .unwrap_or_default();
 
         let events = if options.show_events {
             let events = {
@@ -446,6 +443,11 @@ impl StoredTransaction {
             None
         };
 
+        let raw_effects = options
+            .show_raw_effects
+            .then_some(self.raw_effects)
+            .unwrap_or_default();
+
         Ok(IotaTransactionBlockResponse {
             digest: tx_digest,
             transaction,
@@ -458,7 +460,7 @@ impl StoredTransaction {
             checkpoint: Some(self.checkpoint_sequence_number as u64),
             confirmed_local_execution: None,
             errors: vec![],
-            raw_effects: self.raw_effects,
+            raw_effects,
         })
     }
 

crates/iota-indexer/src/test_utils.rs

@@ -45,6 +45,7 @@ pub async fn start_test_indexer<T: R2D2Connection + Send + 'static>(
     rpc_url: String,
     reader_writer_config: ReaderWriterConfig,
     data_ingestion_path: PathBuf,
+    new_database: Option<&str>,
 ) -> (PgIndexerStore<T>, JoinHandle<Result<(), IndexerError>>) {
     start_test_indexer_impl(
         db_url,
@@ -54,6 +55,7 @@ pub async fn start_test_indexer<T: R2D2Connection + Send + 'static>(
         false,
         Some(data_ingestion_path),
         CancellationToken::new(),
+        new_database,
     )
     .await
 }
@@ -65,17 +67,23 @@ pub async fn start_test_indexer_impl<T: R2D2Connection + 'static>(
     db_url: Option<String>,
     rpc_url: String,
     reader_writer_config: ReaderWriterConfig,
-    reset_database: bool,
+    mut reset_database: bool,
     data_ingestion_path: Option<PathBuf>,
     cancel: CancellationToken,
+    new_database: Option<&str>,
 ) -> (PgIndexerStore<T>, JoinHandle<Result<(), IndexerError>>) {
-    let db_url = db_url.unwrap_or_else(|| {
+    let mut db_url = db_url.unwrap_or_else(|| {
         let pg_host = env::var("POSTGRES_HOST").unwrap_or_else(|_| "localhost".into());
         let pg_port = env::var("POSTGRES_PORT").unwrap_or_else(|_| "32770".into());
         let pw = env::var("POSTGRES_PASSWORD").unwrap_or_else(|_| "postgrespw".into());
         format!("postgres://postgres:{pw}@{pg_host}:{pg_port}")
     });
 
+    if let Some(new_database) = new_database {
+        db_url = replace_db_name(&db_url, new_database).0;
+        reset_database = true;
+    };
+
     let mut config = IndexerConfig {
         db_url: Some(db_url.clone().into()),
         rpc_client_url: rpc_url,