From cc167f6135096a7276eda6795f28829349c61be1 Mon Sep 17 00:00:00 2001 From: Emma Zhong Date: Sat, 19 Oct 2024 17:20:30 -0700 Subject: [PATCH] [indexer][test cluster] support indexer backed rpc in cluster test and add some indexer reader tests (#19906) ## Description This PR adds support in TestCluster to start indexer writer and indexer backed jsonrpc so that any testing done using TestCluster and fullnode jsonrpc can now be easily switched to using indexer jsonrpc. It's a step needed towards deprecation of fullnode jsonrpc. And a few tests are added/adapted from existing rpc tests. ## Test plan Added tests. --- ## Release notes Check each box that your changes affect. If none of the boxes relate to your changes, release notes aren't required. For each box you select, include information after the relevant heading that describes the impact of your changes that a user might notice and any actions they must take to implement updates. - [ ] Protocol: - [ ] Nodes (Validators and Full nodes): - [ ] Indexer: - [ ] JSON-RPC: - [ ] GraphQL: - [ ] CLI: - [ ] Rust SDK: - [ ] REST API: --- Cargo.lock | 4 + crates/sui-cluster-test/src/cluster.rs | 1 + crates/sui-config/src/local_ip_utils.rs | 11 +- crates/sui-indexer/Cargo.toml | 1 + crates/sui-indexer/src/apis/read_api.rs | 6 +- crates/sui-indexer/tests/json_rpc_tests.rs | 141 ++++++++++++++++++ crates/test-cluster/Cargo.toml | 3 + crates/test-cluster/src/indexer_util.rs | 84 +++++++++++ crates/test-cluster/src/lib.rs | 71 +++++++-- .../test-cluster/src/test_indexer_handle.rs | 86 +++++++++++ 10 files changed, 393 insertions(+), 15 deletions(-) create mode 100644 crates/sui-indexer/tests/json_rpc_tests.rs create mode 100644 crates/test-cluster/src/indexer_util.rs create mode 100644 crates/test-cluster/src/test_indexer_handle.rs diff --git a/Cargo.lock b/Cargo.lock index 67ab89cb21ce7..f3f50769be42b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -13844,6 +13844,7 @@ dependencies = [ "sui-sdk", "sui-snapshot", "sui-storage", + "sui-swarm-config", "sui-synthetic-ingestion", "sui-test-transaction-builder", "sui-transaction-builder", @@ -15863,6 +15864,7 @@ dependencies = [ "sui-config", "sui-core", "sui-framework", + "sui-indexer", "sui-json-rpc", "sui-json-rpc-api", "sui-json-rpc-types", @@ -15876,7 +15878,9 @@ dependencies = [ "sui-swarm-config", "sui-test-transaction-builder", "sui-types", + "tempfile", "tokio", + "tokio-util 0.7.10", "tracing", ] diff --git a/crates/sui-cluster-test/src/cluster.rs b/crates/sui-cluster-test/src/cluster.rs index decf58e81714d..c4b27b4afb7f8 100644 --- a/crates/sui-cluster-test/src/cluster.rs +++ b/crates/sui-cluster-test/src/cluster.rs @@ -223,6 +223,7 @@ impl Cluster for LocalNewCluster { // This cluster has fullnode handle, safe to unwrap let fullnode_url = test_cluster.fullnode_handle.rpc_url.clone(); + // TODO: with TestCluster supporting indexer backed rpc as well, we can remove the indexer related logic here. let mut cancellation_tokens = vec![]; let (database, indexer_url, graphql_url) = if options.with_indexer_and_graphql { let database = TempDb::new()?; diff --git a/crates/sui-config/src/local_ip_utils.rs b/crates/sui-config/src/local_ip_utils.rs index 5e7d1298f3629..e8fb02c7f145f 100644 --- a/crates/sui-config/src/local_ip_utils.rs +++ b/crates/sui-config/src/local_ip_utils.rs @@ -122,15 +122,18 @@ pub fn new_udp_address_for_testing(host: &str) -> Multiaddr { .unwrap() } -/// Returns a new unique TCP address (SocketAddr) for localhost, by finding a new available port on localhost. -pub fn new_local_tcp_socket_for_testing() -> SocketAddr { +/// Returns a new unique TCP address in String format for localhost, by finding a new available port on localhost. +pub fn new_local_tcp_socket_for_testing_string() -> String { format!( "{}:{}", localhost_for_testing(), get_available_port(&localhost_for_testing()) ) - .parse() - .unwrap() +} + +/// Returns a new unique TCP address (SocketAddr) for localhost, by finding a new available port on localhost. +pub fn new_local_tcp_socket_for_testing() -> SocketAddr { + new_local_tcp_socket_for_testing_string().parse().unwrap() } /// Returns a new unique TCP address (Multiaddr) for localhost, by finding a new available port on localhost. diff --git a/crates/sui-indexer/Cargo.toml b/crates/sui-indexer/Cargo.toml index 3470488d510f1..0b22f81ff2bba 100644 --- a/crates/sui-indexer/Cargo.toml +++ b/crates/sui-indexer/Cargo.toml @@ -78,6 +78,7 @@ dashmap.workspace = true [dev-dependencies] sui-keys.workspace = true sui-move-build.workspace = true +sui-swarm-config.workspace = true sui-test-transaction-builder.workspace = true test-cluster.workspace = true ntest.workspace = true diff --git a/crates/sui-indexer/src/apis/read_api.rs b/crates/sui-indexer/src/apis/read_api.rs index 78b8715e16ce7..3e3de5343869d 100644 --- a/crates/sui-indexer/src/apis/read_api.rs +++ b/crates/sui-indexer/src/apis/read_api.rs @@ -87,7 +87,11 @@ impl ReadApiServer for ReadApi { object_read_to_object_response(&self.inner, object_read, options.clone()).await }); - futures::future::try_join_all(futures).await + let mut objects = futures::future::try_join_all(futures).await?; + // Resort the objects by the order of the object id. + objects.sort_by_key(|obj| obj.data.as_ref().map(|data| data.object_id)); + + Ok(objects) } async fn get_total_transaction_blocks(&self) -> RpcResult> { diff --git a/crates/sui-indexer/tests/json_rpc_tests.rs b/crates/sui-indexer/tests/json_rpc_tests.rs new file mode 100644 index 0000000000000..6d50c07c04b80 --- /dev/null +++ b/crates/sui-indexer/tests/json_rpc_tests.rs @@ -0,0 +1,141 @@ +// Copyright (c) Mysten Labs, Inc. +// SPDX-License-Identifier: Apache-2.0 + +use sui_json_rpc_api::{CoinReadApiClient, IndexerApiClient, ReadApiClient}; +use sui_json_rpc_types::{ + CoinPage, SuiObjectDataOptions, SuiObjectResponse, SuiObjectResponseQuery, +}; +use sui_swarm_config::genesis_config::DEFAULT_GAS_AMOUNT; +use test_cluster::TestClusterBuilder; + +#[tokio::test] +async fn test_get_owned_objects() -> Result<(), anyhow::Error> { + let cluster = TestClusterBuilder::new() + .with_indexer_backed_rpc() + .build() + .await; + + let http_client = cluster.rpc_client(); + let address = cluster.get_address_0(); + + let data_option = SuiObjectDataOptions::new().with_owner(); + let objects = http_client + .get_owned_objects( + address, + Some(SuiObjectResponseQuery::new_with_options( + data_option.clone(), + )), + None, + None, + ) + .await? + .data; + let fullnode_objects = cluster + .fullnode_handle + .rpc_client + .get_owned_objects( + address, + Some(SuiObjectResponseQuery::new_with_options( + data_option.clone(), + )), + None, + None, + ) + .await? + .data; + assert_eq!(5, objects.len()); + // TODO: right now we compare the results from indexer and fullnode, but as we deprecate fullnode rpc, + // we should change this to compare the results with the object id/digest from genesis potentially. + assert_eq!(objects, fullnode_objects); + + for obj in &objects { + let oref = obj.clone().into_object().unwrap(); + let result = http_client + .get_object(oref.object_id, Some(data_option.clone())) + .await?; + assert!( + matches!(result, SuiObjectResponse { data: Some(object), .. } if oref.object_id == object.object_id && object.owner.unwrap().get_owner_address()? == address) + ); + } + + // Multiget objectIDs test + let object_ids: Vec<_> = objects + .iter() + .map(|o| o.object().unwrap().object_id) + .collect(); + + let object_resp = http_client + .multi_get_objects(object_ids.clone(), None) + .await?; + let fullnode_object_resp = cluster + .fullnode_handle + .rpc_client + .multi_get_objects(object_ids, None) + .await?; + assert_eq!(5, object_resp.len()); + // TODO: right now we compare the results from indexer and fullnode, but as we deprecate fullnode rpc, + // we should change this to compare the results with the object id/digest from genesis potentially. + assert_eq!(object_resp, fullnode_object_resp); + Ok(()) +} + +#[tokio::test] +async fn test_get_coins() -> Result<(), anyhow::Error> { + let cluster = TestClusterBuilder::new() + .with_indexer_backed_rpc() + .build() + .await; + let http_client = cluster.rpc_client(); + let address = cluster.get_address_0(); + + let result: CoinPage = http_client.get_coins(address, None, None, None).await?; + assert_eq!(5, result.data.len()); + assert!(!result.has_next_page); + + // We should get 0 coins for a non-existent coin type. + let result: CoinPage = http_client + .get_coins(address, Some("0x2::sui::TestCoin".into()), None, None) + .await?; + assert_eq!(0, result.data.len()); + + // We should get all the 5 coins for SUI with the right balance. + let result: CoinPage = http_client + .get_coins(address, Some("0x2::sui::SUI".into()), None, None) + .await?; + assert_eq!(5, result.data.len()); + assert_eq!(result.data[0].balance, DEFAULT_GAS_AMOUNT); + assert!(!result.has_next_page); + + // When we have more than 3 coins, we should get a next page. + let result: CoinPage = http_client + .get_coins(address, Some("0x2::sui::SUI".into()), None, Some(3)) + .await?; + assert_eq!(3, result.data.len()); + assert!(result.has_next_page); + + // We should get the remaining 2 coins with the next page. + let result: CoinPage = http_client + .get_coins( + address, + Some("0x2::sui::SUI".into()), + result.next_cursor, + Some(3), + ) + .await?; + assert_eq!(2, result.data.len(), "{:?}", result); + assert!(!result.has_next_page); + + // No more coins after the last page. + let result: CoinPage = http_client + .get_coins( + address, + Some("0x2::sui::SUI".into()), + result.next_cursor, + None, + ) + .await?; + assert_eq!(0, result.data.len(), "{:?}", result); + assert!(!result.has_next_page); + + Ok(()) +} diff --git a/crates/test-cluster/Cargo.toml b/crates/test-cluster/Cargo.toml index 58ca4d37edd93..70bb3d0be5023 100644 --- a/crates/test-cluster/Cargo.toml +++ b/crates/test-cluster/Cargo.toml @@ -17,11 +17,14 @@ futures.workspace = true tracing.workspace = true jsonrpsee.workspace = true tokio = { workspace = true, features = ["full", "tracing", "test-util"] } +tokio-util.workspace = true rand.workspace = true +tempfile.workspace = true sui-config.workspace = true sui-core = { workspace = true, features = ["test-utils"] } sui-framework.workspace = true sui-swarm-config.workspace = true +sui-indexer.workspace = true sui-json-rpc.workspace = true sui-json-rpc-types.workspace = true sui-json-rpc-api.workspace = true diff --git a/crates/test-cluster/src/indexer_util.rs b/crates/test-cluster/src/indexer_util.rs new file mode 100644 index 0000000000000..5d3b8b5605dd7 --- /dev/null +++ b/crates/test-cluster/src/indexer_util.rs @@ -0,0 +1,84 @@ +// Copyright (c) Mysten Labs, Inc. +// SPDX-License-Identifier: Apache-2.0 + +use jsonrpsee::http_client::{HttpClient, HttpClientBuilder}; +use std::path::PathBuf; +use std::time::Duration; +use sui_config::local_ip_utils::get_available_port; +use sui_indexer::tempdb::TempDb; +use sui_indexer::test_utils::{ + start_indexer_jsonrpc_for_testing, start_indexer_writer_for_testing, +}; +use sui_json_rpc_api::ReadApiClient; +use sui_sdk::{SuiClient, SuiClientBuilder}; +use tempfile::TempDir; +use tokio::time::sleep; + +pub(crate) struct IndexerHandle { + pub(crate) rpc_client: HttpClient, + pub(crate) sui_client: SuiClient, + pub(crate) rpc_url: String, + #[allow(unused)] + cancellation_tokens: Vec, + #[allow(unused)] + data_ingestion_dir: Option, + #[allow(unused)] + database: TempDb, +} + +// TODO: this only starts indexer writer and reader (jsonrpc server) today. +// Consider adding graphql server here as well. +pub(crate) async fn setup_indexer_backed_rpc( + fullnode_rpc_url: String, + temp_data_ingestion_dir: Option, + data_ingestion_path: PathBuf, +) -> IndexerHandle { + let mut cancellation_tokens = vec![]; + let database = TempDb::new().unwrap(); + let pg_address = database.database().url().as_str().to_owned(); + let indexer_jsonrpc_address = format!("127.0.0.1:{}", get_available_port("127.0.0.1")); + + // Start indexer writer + let (_, _, writer_token) = start_indexer_writer_for_testing( + pg_address.clone(), + None, + None, + Some(data_ingestion_path.clone()), + None, + ) + .await; + cancellation_tokens.push(writer_token.drop_guard()); + + // Start indexer jsonrpc service + let (_, reader_token) = start_indexer_jsonrpc_for_testing( + pg_address.clone(), + fullnode_rpc_url, + indexer_jsonrpc_address.clone(), + None, + ) + .await; + cancellation_tokens.push(reader_token.drop_guard()); + + let rpc_address = format!("http://{}", indexer_jsonrpc_address); + + let rpc_client = HttpClientBuilder::default().build(&rpc_address).unwrap(); + + // Wait for the rpc client to be ready + while rpc_client.get_chain_identifier().await.is_err() { + sleep(Duration::from_millis(100)).await; + } + + let sui_client = SuiClientBuilder::default() + .build(&rpc_address) + .await + .unwrap(); + + IndexerHandle { + rpc_client, + sui_client, + rpc_url: rpc_address.clone(), + database, + data_ingestion_dir: temp_data_ingestion_dir, + cancellation_tokens, + } +} diff --git a/crates/test-cluster/src/lib.rs b/crates/test-cluster/src/lib.rs index 05643d8716b01..5cfaf7d2b2ad8 100644 --- a/crates/test-cluster/src/lib.rs +++ b/crates/test-cluster/src/lib.rs @@ -61,6 +61,8 @@ use tokio::time::{timeout, Instant}; use tokio::{task::JoinHandle, time::sleep}; use tracing::{error, info}; +mod test_indexer_handle; + const NUM_VALIDATOR: usize = 4; pub struct FullNodeHandle { @@ -90,23 +92,33 @@ pub struct TestCluster { pub swarm: Swarm, pub wallet: WalletContext, pub fullnode_handle: FullNodeHandle, + indexer_handle: Option, } impl TestCluster { pub fn rpc_client(&self) -> &HttpClient { - &self.fullnode_handle.rpc_client + self.indexer_handle + .as_ref() + .map(|h| &h.rpc_client) + .unwrap_or(&self.fullnode_handle.rpc_client) } pub fn sui_client(&self) -> &SuiClient { - &self.fullnode_handle.sui_client + self.indexer_handle + .as_ref() + .map(|h| &h.sui_client) + .unwrap_or(&self.fullnode_handle.sui_client) } - pub fn quorum_driver_api(&self) -> &QuorumDriverApi { - self.sui_client().quorum_driver_api() + pub fn rpc_url(&self) -> &str { + self.indexer_handle + .as_ref() + .map(|h| h.rpc_url.as_str()) + .unwrap_or(&self.fullnode_handle.rpc_url) } - pub fn rpc_url(&self) -> &str { - &self.fullnode_handle.rpc_url + pub fn quorum_driver_api(&self) -> &QuorumDriverApi { + self.sui_client().quorum_driver_api() } pub fn wallet(&mut self) -> &WalletContext { @@ -829,6 +841,8 @@ pub struct TestClusterBuilder { max_submit_position: Option, submit_delay_step_override_millis: Option, validator_state_accumulator_v2_enabled_config: StateAccumulatorV2EnabledConfig, + + indexer_backed_rpc: bool, } impl TestClusterBuilder { @@ -859,6 +873,7 @@ impl TestClusterBuilder { validator_state_accumulator_v2_enabled_config: StateAccumulatorV2EnabledConfig::Global( true, ), + indexer_backed_rpc: false, } } @@ -1057,6 +1072,11 @@ impl TestClusterBuilder { self } + pub fn with_indexer_backed_rpc(mut self) -> Self { + self.indexer_backed_rpc = true; + self + } + pub async fn build(mut self) -> TestCluster { // All test clusters receive a continuous stream of random JWKs. // If we later use zklogin authenticated transactions in tests we will need to supply @@ -1087,20 +1107,50 @@ impl TestClusterBuilder { })); } + let mut temp_data_ingestion_dir = None; + let mut data_ingestion_path = None; + + if self.indexer_backed_rpc { + if self.data_ingestion_dir.is_none() { + temp_data_ingestion_dir = Some(tempfile::tempdir().unwrap()); + self.data_ingestion_dir = Some( + temp_data_ingestion_dir + .as_ref() + .unwrap() + .path() + .to_path_buf(), + ); + assert!(self.data_ingestion_dir.is_some()); + } + assert!(self.data_ingestion_dir.is_some()); + data_ingestion_path = Some(self.data_ingestion_dir.as_ref().unwrap().to_path_buf()); + } + let swarm = self.start_swarm().await.unwrap(); let working_dir = swarm.dir(); - let mut wallet_conf: SuiClientConfig = - PersistedConfig::read(&working_dir.join(SUI_CLIENT_CONFIG)).unwrap(); - let fullnode = swarm.fullnodes().next().unwrap(); let json_rpc_address = fullnode.config().json_rpc_address; let fullnode_handle = FullNodeHandle::new(fullnode.get_node_handle().unwrap(), json_rpc_address).await; + let (rpc_url, indexer_handle) = if self.indexer_backed_rpc { + let handle = test_indexer_handle::IndexerHandle::new( + fullnode_handle.rpc_url.clone(), + temp_data_ingestion_dir, + data_ingestion_path.unwrap(), + ) + .await; + (handle.rpc_url.clone(), Some(handle)) + } else { + (fullnode_handle.rpc_url.clone(), None) + }; + + let mut wallet_conf: SuiClientConfig = + PersistedConfig::read(&working_dir.join(SUI_CLIENT_CONFIG)).unwrap(); wallet_conf.envs.push(SuiEnv { alias: "localnet".to_string(), - rpc: fullnode_handle.rpc_url.clone(), + rpc: rpc_url, ws: None, basic_auth: None, }); @@ -1118,6 +1168,7 @@ impl TestClusterBuilder { swarm, wallet, fullnode_handle, + indexer_handle, } } diff --git a/crates/test-cluster/src/test_indexer_handle.rs b/crates/test-cluster/src/test_indexer_handle.rs new file mode 100644 index 0000000000000..6e0595fe567a9 --- /dev/null +++ b/crates/test-cluster/src/test_indexer_handle.rs @@ -0,0 +1,86 @@ +// Copyright (c) Mysten Labs, Inc. +// SPDX-License-Identifier: Apache-2.0 + +use jsonrpsee::http_client::{HttpClient, HttpClientBuilder}; +use std::path::PathBuf; +use std::time::Duration; +use sui_config::local_ip_utils::new_local_tcp_socket_for_testing_string; +use sui_indexer::tempdb::TempDb; +use sui_indexer::test_utils::{ + start_indexer_jsonrpc_for_testing, start_indexer_writer_for_testing, +}; +use sui_json_rpc_api::ReadApiClient; +use sui_sdk::{SuiClient, SuiClientBuilder}; +use tempfile::TempDir; +use tokio::time::sleep; + +pub(crate) struct IndexerHandle { + pub(crate) rpc_client: HttpClient, + pub(crate) sui_client: SuiClient, + pub(crate) rpc_url: String, + #[allow(unused)] + cancellation_tokens: Vec, + #[allow(unused)] + data_ingestion_dir: Option, + #[allow(unused)] + database: TempDb, +} + +impl IndexerHandle { + // TODO: this only starts indexer writer and reader (jsonrpc server) today. + // Consider adding graphql server here as well. + pub async fn new( + fullnode_rpc_url: String, + data_ingestion_dir: Option, + data_ingestion_path: PathBuf, + ) -> IndexerHandle { + let mut cancellation_tokens = vec![]; + let database = TempDb::new().unwrap(); + let pg_address = database.database().url().as_str().to_owned(); + let indexer_jsonrpc_address = new_local_tcp_socket_for_testing_string(); + + // Start indexer writer + let (_, _, writer_token) = start_indexer_writer_for_testing( + pg_address.clone(), + None, + None, + Some(data_ingestion_path.clone()), + None, + ) + .await; + cancellation_tokens.push(writer_token.drop_guard()); + + // Start indexer jsonrpc service + let (_, reader_token) = start_indexer_jsonrpc_for_testing( + pg_address.clone(), + fullnode_rpc_url, + indexer_jsonrpc_address.clone(), + None, + ) + .await; + cancellation_tokens.push(reader_token.drop_guard()); + + let rpc_address = format!("http://{}", indexer_jsonrpc_address); + + let rpc_client = HttpClientBuilder::default().build(&rpc_address).unwrap(); + + // Wait for the rpc client to be ready + while rpc_client.get_chain_identifier().await.is_err() { + sleep(Duration::from_millis(100)).await; + } + + let sui_client = SuiClientBuilder::default() + .build(&rpc_address) + .await + .unwrap(); + + IndexerHandle { + rpc_client, + sui_client, + rpc_url: rpc_address.clone(), + database, + data_ingestion_dir, + cancellation_tokens, + } + } +}