refactor(indexer): Move state_changes from StreamerMessage root to IndexerShard instead

CODEOWNERS

@@ -8,6 +8,7 @@
 /chain/jsonrpc/ @khorolets @rtsai123
 /chain/jsonrpc-primitives/ @khorolets @miraclx @rtsai123
 /chain/indexer/ @khorolets
+/chain/indexer-primitives/ @khorolets
 /chain/rosetta-rpc/ @frol @mina86
 /utils @pmnoxx @mm-near
 

Cargo.toml

@@ -22,6 +22,7 @@ members = [
     "chain/client-primitives",
     "chain/network",
     "chain/indexer",
+    "chain/indexer-primitives",
     "chain/jsonrpc",
     "chain/jsonrpc/fuzz",
     "chain/jsonrpc/client",

chain/client-primitives/src/types.rs

@@ -13,7 +13,7 @@ use near_primitives::hash::CryptoHash;
 use near_primitives::merkle::{MerklePath, PartialMerkleTree};
 use near_primitives::sharding::ChunkHash;
 use near_primitives::types::{
-    AccountId, BlockHeight, BlockReference, EpochReference, MaybeBlockId, ShardId,
+    AccountId, BlockHeight, BlockReference, EpochId, EpochReference, MaybeBlockId, ShardId,
     TransactionOrReceiptId,
 };
 use near_primitives::utils::generate_random_string;
@@ -626,6 +626,15 @@ impl Message for GetStateChangesWithCauseInBlock {
     type Result = Result<StateChangesView, GetStateChangesError>;
 }
 
+pub struct GetStateChangesWithCauseInBlockForTrackedShards {
+    pub block_hash: CryptoHash,
+    pub epoch_id: EpochId,
+}
+
+impl Message for GetStateChangesWithCauseInBlockForTrackedShards {
+    type Result = Result<HashMap<ShardId, StateChangesView>, GetStateChangesError>;
+}
+
 pub struct GetExecutionOutcome {
     pub id: TransactionOrReceiptId,
 }

chain/client/src/lib.rs

@@ -2,9 +2,9 @@ pub use near_client_primitives::types::{
     Error, GetBlock, GetBlockHash, GetBlockProof, GetBlockProofResponse, GetBlockWithMerkleTree,
     GetChunk, GetExecutionOutcome, GetExecutionOutcomeResponse, GetExecutionOutcomesForBlock,
     GetGasPrice, GetNetworkInfo, GetNextLightClientBlock, GetProtocolConfig, GetReceipt,
-    GetStateChanges, GetStateChangesInBlock, GetStateChangesWithCauseInBlock, GetValidatorInfo,
-    GetValidatorOrdered, Query, QueryError, Status, StatusResponse, SyncStatus, TxStatus,
-    TxStatusError,
+    GetStateChanges, GetStateChangesInBlock, GetStateChangesWithCauseInBlock,
+    GetStateChangesWithCauseInBlockForTrackedShards, GetValidatorInfo, GetValidatorOrdered, Query,
+    QueryError, Status, StatusResponse, SyncStatus, TxStatus, TxStatusError,
 };
 
 pub use crate::client::Client;

chain/client/src/view_client.rs

@@ -22,8 +22,9 @@ use near_client_primitives::types::{
     GetBlockProofResponse, GetBlockWithMerkleTree, GetChunkError, GetExecutionOutcome,
     GetExecutionOutcomeError, GetExecutionOutcomesForBlock, GetGasPrice, GetGasPriceError,
     GetNextLightClientBlockError, GetProtocolConfig, GetProtocolConfigError, GetReceipt,
-    GetReceiptError, GetStateChangesError, GetStateChangesWithCauseInBlock, GetValidatorInfoError,
-    Query, QueryError, TxStatus, TxStatusError,
+    GetReceiptError, GetStateChangesError, GetStateChangesWithCauseInBlock,
+    GetStateChangesWithCauseInBlockForTrackedShards, GetValidatorInfoError, Query, QueryError,
+    TxStatus, TxStatusError,
 };
 use near_network::types::{NetworkRequests, PeerManagerAdapter, PeerManagerMessageRequest};
 #[cfg(feature = "test_features")]
@@ -783,6 +784,43 @@ impl Handler<GetStateChangesWithCauseInBlock> for ViewClientActor {
     }
 }
 
+/// Returns a hashmap where the key represents the ShardID and the value
+/// is the list of changes in a store with causes for a given block.
+impl Handler<GetStateChangesWithCauseInBlockForTrackedShards> for ViewClientActor {
+    type Result = Result<HashMap<ShardId, StateChangesView>, GetStateChangesError>;
+
+    #[perf]
+    fn handle(
+        &mut self,
+        msg: GetStateChangesWithCauseInBlockForTrackedShards,
+        _: &mut Self::Context,
+    ) -> Self::Result {
+        let state_changes_with_cause_in_block =
+            self.chain.store().get_state_changes_with_cause_in_block(&msg.block_hash)?;
+
+        let mut state_changes_with_cause_split_by_shard_id: HashMap<ShardId, StateChangesView> =
+            HashMap::new();
+        for state_change_with_cause in state_changes_with_cause_in_block {
+            let account_id = state_change_with_cause.value.affected_account_id();
+            let shard_id = match self
+                .runtime_adapter
+                .account_id_to_shard_id(account_id, &msg.epoch_id)
+            {
+                Ok(shard_id) => shard_id,
+                Err(err) => {
+                    return Err(GetStateChangesError::IOError { error_message: format!("{}", err) })
+                }
+            };
+
+            let state_changes =
+                state_changes_with_cause_split_by_shard_id.entry(shard_id).or_default();
+            state_changes.push(state_change_with_cause.into());
+        }
+
+        Ok(state_changes_with_cause_split_by_shard_id)
+    }
+}
+
 /// Returns the next light client block, given the hash of the last block known to the light client.
 /// There are three cases:
 ///  1. The last block known to the light client is in the same epoch as the tip:

chain/indexer-primitives/Cargo.toml

@@ -0,0 +1,19 @@
+[package]
+name = "near-indexer-primitives"
+version = "0.0.0"
+authors = ["Near Inc <hello@nearprotocol.com>"]
+publish = true
+# Please update rust-toolchain.toml as well when changing version here:
+rust-version = "1.56.0"
+edition = "2021"
+license = "MIT OR Apache-2.0"
+repository = "https://github.com/near/nearcore"
+description = "This crate hosts structures for the NEAR Indexer Framework types"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+serde = { version = "1", features = [ "derive" ] }
+serde_json = "1.0.55"
+
+near-primitives = { path = "../../core/primitives" }

chain/indexer-primitives/LICENSE-APACHE

@@ -0,0 +1 @@
+../../licenses/LICENSE-APACHE

chain/indexer-primitives/LICENSE-MIT

@@ -0,0 +1 @@
+../../licenses/LICENSE-MIT

chain/indexer-primitives/README.md

@@ -0,0 +1,3 @@
+# near-indexer-primitives
+
+This crate holds the types that is used in NEAR Indexer Framework to allow other projects to use them without a need to depend on entire `nearcore`.

chain/indexer/src/streamer/types.rs → chain/indexer-primitives/src/lib.rs

@@ -8,7 +8,6 @@ pub use near_primitives::{types, views};
 pub struct StreamerMessage {
     pub block: views::BlockView,
     pub shards: Vec<IndexerShard>,
-    pub state_changes: views::StateChangesView,
 }
 
 #[derive(Debug, Serialize, Deserialize)]
@@ -42,4 +41,5 @@ pub struct IndexerShard {
     pub shard_id: types::ShardId,
     pub chunk: Option<IndexerChunkView>,
     pub receipt_execution_outcomes: Vec<IndexerExecutionOutcomeWithReceipt>,
+    pub state_changes: views::StateChangesView,
 }

chain/indexer/CHANGELOG.md

@@ -1,5 +1,17 @@
 # Changelog
 
+## 1.26.x (UNRELEASED)
+
+* `state_changes` field is moved from the top-level `StreamerMessage` to `IndexerShard` struct to align better with the sharded nature of NEAR protocol. In the future, when nearcore will be able to track only a subset of shards, this API will work naturally, so we take pro-active measures to solidify the APIs
+* All the NEAR Indexer Framework types were extracted to a separate crate `near-indexer-primitives`
+* Increase the streamer size from 16 to 100 in order to increase the speed of streaming messages (affects reindexing jobs)
+
+## Breaking changes
+
+The field `state_changes` is moved from the root of `StreamerMessage`
+to the `IndexerShard.state_changes` and now contains only changes related
+to the specific shard.
+
 ## 0.10.1
 
 * (mainnet only) Add additional handler to inject restored receipts to the block #47317863. See [PR 4248](https://github.com/near/nearcore/pull/4248) for reference
@@ -25,10 +37,10 @@ Since the change of reading genesis method to optimize memory usage. You'd be ab
 ## 0.9.0 (do not use this version, it contains a bug)
 
 * Introduce `IndexerShard` structure which contains corresponding chunks and `IndexerExecutionOutcomeWithReceipt`
-* `receipt` field in `IndexerExecutionOutcomeWithReceipt` is no longer optional as it used to be always set anyway, 
-so now we explicitly communicate this relation ("every outcome has a corresponding receipt") through the type system
+* `receipt` field in `IndexerExecutionOutcomeWithReceipt` is no longer optional as it used to be always set anyway,
+  so now we explicitly communicate this relation ("every outcome has a corresponding receipt") through the type system
 * Introduce `IndexerExecutionOutcomeWithOptionalReceipt` which is the same as `IndexerExecutionOutcomeWithReceipt`
-but with optional `receipt` field.
+  but with optional `receipt` field.
 
 ## Breaking changes
 
@@ -38,7 +50,7 @@ but with optional `receipt` field.
   StreamerMessage {
     block: BlockView,
     shards: Vec<IndexerShard>,
-    state_changes: StateChangesView,  
+    state_changes: StateChangesView,
   }
   ```
 
@@ -64,7 +76,7 @@ created and started on the Indexer implementation, not on the Indexer Framework
 
 ## Breaking changes
 
-* `StreamerMessage` now contains `StateChangesView` which is an alias for  `Vec<StateChangesWithCauseView>`, previously it contained `StateChangesKindsView`
+* `StreamerMessage` now contains `StateChangesView` which is an alias for `Vec<StateChangesWithCauseView>`, previously it contained `StateChangesKindsView`
 
 ## 0.6.0
 
@@ -73,8 +85,8 @@ created and started on the Indexer implementation, not on the Indexer Framework
 ## Breaking changes
 
 * `IndexerConfig` was extended with another field `await_for_node_synced`. Corresponding enum is `AwaitForNodeSyncedEnum` with variants:
-  * `WaitForFullSync` - await for node to be fully synced (previous default behaviour)
-  * `StreamWhileSyncing`- start streaming right away while node is syncing (it's useful in case of Indexing from genesis)
+  - `WaitForFullSync` - await for node to be fully synced (previous default behaviour)
+  - `StreamWhileSyncing`- start streaming right away while node is syncing (it's useful in case of Indexing from genesis)
 
 ## 0.5.0
 
@@ -83,6 +95,7 @@ created and started on the Indexer implementation, not on the Indexer Framework
 ## Breaking changes
 
 Since #3529 nearcore stores `ExecutionOutcome`s in their execution order, and we can also attribute outcomes to specific chunk. That's why:
+
 * `receipt_execution_outcomes` was moved from `StreamerMessage` to a relevant `IndexerChunkView`
 * `ExecutionOutcomesWithReceipts` type alias was removed (just use `Vec<IndexerExecutionOutcomeWithReceipt>` instead)
 
@@ -93,20 +106,19 @@ Since #3529 nearcore stores `ExecutionOutcome`s in their execution order, and we
 ## Breaking changes
 
 * For local receipt to have a relation to specific chunk we have prepended them to original receipts in particular chunk
-as in the most cases local receipts are executed before normal receipts. That's why there is no reason to have `local_receipts`
-field in `StreamerMessage` struct anymore. `local_receipts` field was removed.
+  as in the most cases local receipts are executed before normal receipts. That's why there is no reason to have `local_receipts`
+  field in `StreamerMessage` struct anymore. `local_receipts` field was removed.
 
 ## 0.3.1
 
 * Add local receipt to `receipt_execution_outcomes` if possible
 
 ## 0.3.0
 
-
 ### Breaking changes
 
 * To extended the `receipt_execution_outcomes` with information about the corresponding receipt we had to break the API
-(the old outcome structure is just one layer deeper now [under `execution_outcome` field])
+  (the old outcome structure is just one layer deeper now [under `execution_outcome` field])
 
 ## 0.2.0
 

chain/indexer/Cargo.toml

@@ -22,5 +22,6 @@ nearcore = { path = "../../nearcore" }
 near-client = { path = "../client" }
 near-chain-configs = { path = "../../core/chain-configs" }
 near-crypto = { path = "../../core/crypto" }
+near-indexer-primitives = { path = "../indexer-primitives" }
 near-primitives = { path = "../../core/primitives" }
 node-runtime = { path = "../../runtime/runtime" }

chain/indexer/src/lib.rs

@@ -3,16 +3,16 @@
 use anyhow::Context;
 use tokio::sync::mpsc;
 
+use near_chain_configs::GenesisValidationMode;
 pub use near_primitives;
 use near_primitives::types::Gas;
 pub use nearcore::{get_default_home, init_configs, NearConfig};
 
-pub use self::streamer::{
+pub use near_indexer_primitives::{
     IndexerChunkView, IndexerExecutionOutcomeWithOptionalReceipt,
     IndexerExecutionOutcomeWithReceipt, IndexerShard, IndexerTransactionWithOutcome,
     StreamerMessage,
 };
-use near_chain_configs::GenesisValidationMode;
 
 mod streamer;
 
@@ -113,8 +113,8 @@ impl Indexer {
     }
 
     /// Boots up `near_indexer::streamer`, so it monitors the new blocks with chunks, transactions, receipts, and execution outcomes inside. The returned stream handler should be drained and handled on the user side.
-    pub fn streamer(&self) -> mpsc::Receiver<streamer::StreamerMessage> {
-        let (sender, receiver) = mpsc::channel(16);
+    pub fn streamer(&self) -> mpsc::Receiver<StreamerMessage> {
+        let (sender, receiver) = mpsc::channel(100);
         actix::spawn(streamer::start(
             self.view_client.clone(),
             self.client.clone(),

chain/indexer/src/streamer/fetchers.rs

@@ -6,11 +6,11 @@ use actix::Addr;
 use futures::stream::StreamExt;
 use tracing::warn;
 
-pub use near_primitives::hash::CryptoHash;
-pub use near_primitives::{types, views};
+use near_indexer_primitives::IndexerExecutionOutcomeWithOptionalReceipt;
+use near_primitives::hash::CryptoHash;
+use near_primitives::{types, views};
 
 use super::errors::FailedToFetchData;
-use super::types::IndexerExecutionOutcomeWithOptionalReceipt;
 use super::INDEXER;
 
 pub(crate) async fn fetch_status(
@@ -60,9 +60,10 @@ pub(crate) async fn fetch_block_by_hash(
 pub(crate) async fn fetch_state_changes(
     client: &Addr<near_client::ViewClientActor>,
     block_hash: CryptoHash,
-) -> Result<views::StateChangesView, FailedToFetchData> {
+    epoch_id: near_primitives::types::EpochId,
+) -> Result<HashMap<near_primitives::types::ShardId, views::StateChangesView>, FailedToFetchData> {
     client
-        .send(near_client::GetStateChangesWithCauseInBlock { block_hash })
+        .send(near_client::GetStateChangesWithCauseInBlockForTrackedShards { block_hash, epoch_id })
         .await?
         .map_err(|err| FailedToFetchData::String(err.to_string()))
 }