Chain index integration

integration-tests/Cargo.toml

@@ -31,6 +31,7 @@ serde_json = { workspace = true, features = ["preserve_order"] }
 futures.workspace = true
 tempfile.workspace = true
 tower = { workspace = true, features = ["buffer", "util"] }
+hex = { workspace = true }
 
 # Runtime
 tokio = { workspace = true }

integration-tests/tests/chain_cache.rs

@@ -176,7 +176,7 @@ mod chain_query_interface {
                 )
                 .await
                 .unwrap();
-                let index_reader = chain_index.subscriber().await;
+                let index_reader = chain_index.subscriber();
                 tokio::time::sleep(Duration::from_secs(3)).await;
 
                 (
@@ -209,7 +209,7 @@ mod chain_query_interface {
                 )
                 .await
                 .unwrap();
-                let index_reader = chain_index.subscriber().await;
+                let index_reader = chain_index.subscriber();
                 tokio::time::sleep(Duration::from_secs(3)).await;
 
                 (test_manager, json_service, None, chain_index, index_reader)

integration-tests/tests/fetch_service.rs

@@ -1,12 +1,14 @@
 //! These tests compare the output of `FetchService` with the output of `JsonRpcConnector`.
 
 use futures::StreamExt as _;
+use hex::ToHex as _;
 use zaino_fetch::jsonrpsee::connector::{test_node_and_return_url, JsonRpSeeConnector};
 use zaino_proto::proto::compact_formats::CompactBlock;
 use zaino_proto::proto::service::{
     AddressList, BlockId, BlockRange, GetAddressUtxosArg, GetMempoolTxRequest, GetSubtreeRootsArg,
     PoolType, TransparentAddressBlockFilter, TxFilter,
 };
+use zaino_state::ChainIndex;
 use zaino_state::FetchServiceSubscriber;
 #[allow(deprecated)]
 use zaino_state::{FetchService, LightWalletIndexer, StatusType, ZcashIndexer};
@@ -316,22 +318,30 @@ pub async fn test_get_mempool_info<V: ValidatorExt>(validator: &ValidatorKind) {
     let info = fetch_service_subscriber.get_mempool_info().await.unwrap();
 
     // Derive expected values directly from the current mempool contents.
-    let entries = fetch_service_subscriber.mempool.get_mempool().await;
+
+    let keys = fetch_service_subscriber
+        .indexer
+        .get_mempool_txids()
+        .await
+        .unwrap();
+
+    let values = fetch_service_subscriber
+        .indexer
+        .get_mempool_transactions(Vec::new())
+        .await
+        .unwrap();
 
     // Size
-    assert_eq!(info.size, entries.len() as u64);
+    assert_eq!(info.size, values.len() as u64);
     assert!(info.size >= 1);
 
     // Bytes: sum of SerializedTransaction lengths
-    let expected_bytes: u64 = entries
-        .iter()
-        .map(|(_, value)| value.serialized_tx.as_ref().as_ref().len() as u64)
-        .sum();
+    let expected_bytes: u64 = values.iter().map(|entry| entry.len() as u64).sum();
 
     // Key heap bytes: sum of txid String capacities
-    let expected_key_heap_bytes: u64 = entries
+    let expected_key_heap_bytes: u64 = keys
         .iter()
-        .map(|(key, _)| key.txid.capacity() as u64)
+        .map(|key| key.encode_hex::<String>().capacity() as u64)
         .sum();
 
     let expected_usage = expected_bytes.saturating_add(expected_key_heap_bytes);
@@ -527,12 +537,12 @@ async fn fetch_service_get_address_tx_ids<V: ValidatorExt>(validator: &Validator
         .generate_blocks_and_poll_indexer(1, &fetch_service_subscriber)
         .await;
 
-    let chain_height = fetch_service_subscriber
-        .block_cache
-        .get_chain_height()
-        .await
-        .unwrap()
-        .0;
+    let chain_height: u32 = fetch_service_subscriber
+        .indexer
+        .snapshot_nonfinalized_state()
+        .best_tip
+        .height
+        .into();
     dbg!(&chain_height);
 
     let fetch_service_txids = fetch_service_subscriber
@@ -1168,11 +1178,10 @@ async fn fetch_service_get_block_range_returns_all_pools<V: ValidatorExt>(
 
     assert_eq!(compact_block.height, end_height);
 
-    let expected_transaction_count = if matches!(validator, ValidatorKind::Zebrad) {
-        3
-    } else {
-        4 // zcashd uses shielded coinbase which will add an extra compact tx
-    };
+    // Transparent tx are now included in compact blocks unless specified so the
+    // expected block count should be 4 (3 sent tx + coinbase)
+    let expected_transaction_count = 4;
+
     // the compact block has the right number of transactions
     assert_eq!(compact_block.vtx.len(), expected_transaction_count);
 
@@ -1557,12 +1566,12 @@ async fn fetch_service_get_taddress_txids<V: ValidatorExt>(validator: &Validator
         .generate_blocks_and_poll_indexer(1, &fetch_service_subscriber)
         .await;
 
-    let chain_height = fetch_service_subscriber
-        .block_cache
-        .get_chain_height()
-        .await
-        .unwrap()
-        .0;
+    let chain_height: u32 = fetch_service_subscriber
+        .indexer
+        .snapshot_nonfinalized_state()
+        .best_tip
+        .height
+        .into();
     dbg!(&chain_height);
 
     let block_filter = TransparentAddressBlockFilter {
@@ -1735,8 +1744,11 @@ async fn fetch_service_get_mempool_tx<V: ValidatorExt>(validator: &ValidatorKind
     let mut sorted_fetch_mempool_tx = fetch_mempool_tx.clone();
     sorted_fetch_mempool_tx.sort_by_key(|tx| tx.txid.clone());
 
+    // Transaction IDs from quick_send are already in internal byte order,
+    // which matches what the mempool returns, so no reversal needed
     let tx1_bytes = *tx_1.first().as_ref();
     let tx2_bytes = *tx_2.first().as_ref();
+
     let mut sorted_txids = [tx1_bytes, tx2_bytes];
     sorted_txids.sort_by_key(|hash| *hash);
 
@@ -2363,7 +2375,7 @@ mod zcashd {
             fetch_service_get_lightd_info::<Zcashd>(&ValidatorKind::Zcashd).await;
         }
 
-        #[tokio::test]
+        #[tokio::test(flavor = "multi_thread")]
         pub(crate) async fn get_network_sol_ps() {
             assert_fetch_service_getnetworksols_matches_rpc::<Zcashd>(&ValidatorKind::Zcashd).await;
         }
@@ -2592,7 +2604,7 @@ mod zebrad {
             fetch_service_get_lightd_info::<Zebrad>(&ValidatorKind::Zebrad).await;
         }
 
-        #[tokio::test]
+        #[tokio::test(flavor = "multi_thread")]
         pub(crate) async fn get_network_sol_ps() {
             assert_fetch_service_getnetworksols_matches_rpc::<Zebrad>(&ValidatorKind::Zebrad).await;
         }

integration-tests/tests/json_server.rs

@@ -5,7 +5,8 @@ use zaino_common::{DatabaseConfig, ServiceConfig, StorageConfig};
 
 #[allow(deprecated)]
 use zaino_state::{
-    FetchService, FetchServiceConfig, FetchServiceSubscriber, ZcashIndexer, ZcashService as _,
+    ChainIndex, FetchService, FetchServiceConfig, FetchServiceSubscriber, ZcashIndexer,
+    ZcashService as _,
 };
 use zaino_testutils::from_inputs;
 use zaino_testutils::{TestManager, ValidatorKind};
@@ -627,11 +628,11 @@ async fn get_address_tx_ids_inner() {
     .await;
 
     let chain_height = zcashd_subscriber
-        .block_cache
-        .get_chain_height()
-        .await
-        .unwrap()
-        .0;
+        .indexer
+        .snapshot_nonfinalized_state()
+        .best_tip
+        .height
+        .into();
     dbg!(&chain_height);
 
     let zcashd_txids = zcashd_subscriber
@@ -836,7 +837,13 @@ mod zcashd {
 
                 assert_eq!(zcashd_mining_info, zaino_mining_info);
 
-                test_manager.local_net.generate_blocks(1).await.unwrap();
+                generate_blocks_and_poll_all_chain_indexes(
+                    1,
+                    &test_manager,
+                    zaino_subscriber.clone(),
+                    zcashd_subscriber.clone(),
+                )
+                .await;
             }
 
             test_manager.close().await;
@@ -917,8 +924,13 @@ mod zcashd {
             const BLOCK_LIMIT: u32 = 10;
 
             for i in 0..BLOCK_LIMIT {
-                test_manager.local_net.generate_blocks(1).await.unwrap();
-                tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
+                generate_blocks_and_poll_all_chain_indexes(
+                    1,
+                    &test_manager,
+                    zaino_subscriber.clone(),
+                    zcashd_subscriber.clone(),
+                )
+                .await;
 
                 let block = zcashd_subscriber
                     .z_get_block(i.to_string(), Some(1))

integration-tests/tests/local_cache.rs

@@ -160,12 +160,12 @@ mod zcashd {
         launch_local_cache::<Zcashd>(&ValidatorKind::Zcashd).await;
     }
 
-    #[tokio::test]
+    #[tokio::test(flavor = "multi_thread")]
     async fn process_100_blocks() {
         launch_local_cache_process_n_block_batches::<Zcashd>(&ValidatorKind::Zcashd, 1).await;
     }
 
-    #[tokio::test]
+    #[tokio::test(flavor = "multi_thread")]
     async fn process_200_blocks() {
         launch_local_cache_process_n_block_batches::<Zcashd>(&ValidatorKind::Zcashd, 2).await;
     }
@@ -182,12 +182,12 @@ mod zebrad {
         launch_local_cache::<Zebrad>(&ValidatorKind::Zebrad).await;
     }
 
-    #[tokio::test]
+    #[tokio::test(flavor = "multi_thread")]
     async fn process_100_blocks() {
         launch_local_cache_process_n_block_batches::<Zebrad>(&ValidatorKind::Zebrad, 1).await;
     }
 
-    #[tokio::test]
+    #[tokio::test(flavor = "multi_thread")]
     async fn process_200_blocks() {
         launch_local_cache_process_n_block_batches::<Zebrad>(&ValidatorKind::Zebrad, 2).await;
     }