[indexer][test cluster] support indexer backed rpc in cluster test an…

…d 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:

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()?;

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.

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

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<BigInt<u64>> {

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(())
+}

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

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<tokio_util::sync::DropGuard>,
+    #[allow(unused)]
+    data_ingestion_dir: Option<TempDir>,
+    #[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<TempDir>,
+    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,
+    }
+}