From 92b3731cb5169f17d79111b384dd798489cca31e Mon Sep 17 00:00:00 2001 From: Stan Bondi Date: Wed, 13 Aug 2025 14:07:37 +0400 Subject: [PATCH] fix(state_sync)!: sync state version --- .../src/p2p/rpc/service_impl.rs | 3 +- .../src/p2p/rpc/state_sync_task.rs | 9 +- crates/p2p/proto/rpc.proto | 24 +- crates/p2p/src/conversions/rpc.rs | 7 +- crates/rpc_state_sync/src/state_sync.rs | 234 ++++++++++-------- .../src/column_families/state_transition.rs | 2 + crates/state_store_rocksdb/src/reader.rs | 6 +- crates/state_store_rocksdb/src/writer.rs | 15 ++ .../src/consensus_models/state_transition.rs | 3 +- crates/validator_node_rpc/src/lib.rs | 2 + 10 files changed, 164 insertions(+), 141 deletions(-) diff --git a/applications/tari_validator_node/src/p2p/rpc/service_impl.rs b/applications/tari_validator_node/src/p2p/rpc/service_impl.rs index d435796c6a..923268685b 100644 --- a/applications/tari_validator_node/src/p2p/rpc/service_impl.rs +++ b/applications/tari_validator_node/src/p2p/rpc/service_impl.rs @@ -54,7 +54,7 @@ use tari_rpc_framework::{Request, Response, RpcStatus, Streaming}; use tari_template_lib::types::{HashParseError, TemplateAddress}; use tari_template_manager::interface::TemplateManagerHandle; use tari_transaction::{Transaction, TransactionId}; -use tari_validator_node_rpc::rpc_service::ValidatorNodeRpcService; +use tari_validator_node_rpc::{rpc_service::ValidatorNodeRpcService, STATE_SYNC_MAX_BATCH_SIZE}; use tokio::{sync::mpsc, task}; use crate::p2p::{ @@ -393,6 +393,7 @@ impl ValidatorNodeRpcSe sender, last_state_transition_for_chain, end_epoch, + STATE_SYNC_MAX_BATCH_SIZE, ) .run(), ); diff --git a/applications/tari_validator_node/src/p2p/rpc/state_sync_task.rs b/applications/tari_validator_node/src/p2p/rpc/state_sync_task.rs index 1472590bda..ffe068181a 100644 --- a/applications/tari_validator_node/src/p2p/rpc/state_sync_task.rs +++ b/applications/tari_validator_node/src/p2p/rpc/state_sync_task.rs @@ -14,8 +14,6 @@ use tokio::sync::mpsc; const LOG_TARGET: &str = "tari::ootle::rpc::sync_task"; -const BATCH_SIZE: usize = 100; - type UpdateBuffer = Vec; pub struct StateSyncTask { @@ -23,6 +21,7 @@ pub struct StateSyncTask { sender: mpsc::Sender>, start_state_transition_id: StateTransitionId, current_epoch: Epoch, + batch_size: usize, } impl StateSyncTask { @@ -31,17 +30,19 @@ impl StateSyncTask { sender: mpsc::Sender>, start_state_transition_id: StateTransitionId, current_epoch: Epoch, + batch_size: usize, ) -> Self { Self { store, sender, start_state_transition_id, current_epoch, + batch_size, } } pub async fn run(mut self) -> Result<(), ()> { - let mut buffer = Vec::with_capacity(BATCH_SIZE); + let mut buffer = Vec::with_capacity(self.batch_size); let mut current_state_transition_id = self.start_state_transition_id; let mut counter = 0usize; loop { @@ -93,7 +94,7 @@ impl StateSyncTask { ) -> Result, StorageError> { self.store.with_read_tx(|tx| { let state_transitions = - StateTransition::get_n_after(tx, BATCH_SIZE, current_state_transition_id, self.current_epoch) + StateTransition::get_n_after(tx, self.batch_size, current_state_transition_id, self.current_epoch) .optional()? .unwrap_or_default(); diff --git a/crates/p2p/proto/rpc.proto b/crates/p2p/proto/rpc.proto index 85cea63d00..c96fecebc9 100644 --- a/crates/p2p/proto/rpc.proto +++ b/crates/p2p/proto/rpc.proto @@ -89,29 +89,6 @@ message GetPeersResponse { repeated tari.ootle.network.PeerIdentityClaim claims = 2; } -message VnStateSyncRequest { - tari.ootle.common.SubstateAddress start_address = 1; - tari.ootle.common.SubstateAddress end_address = 2; - repeated tari.ootle.common.SubstateAddress inventory = 3; -} - -message VnStateSyncResponse { - bytes address = 1; - uint32 version = 2; - bytes substate = 3; - - uint64 created_epoch = 4; - uint64 created_height = 5; - bytes created_block = 6; - bytes created_transaction = 7; - bytes created_justify = 8; - - tari.ootle.common.Epoch destroyed_epoch = 9; - bytes destroyed_block = 10; - bytes destroyed_transaction = 11; - bytes destroyed_justify = 12; -} - message GetSubstateRequest { tari.ootle.transaction.SubstateRequirement substate_requirement = 1; } @@ -244,6 +221,7 @@ message SyncStateResponse { message StateTransition { StateTransitionId id = 1; SubstateUpdate update = 2; + uint64 state_version = 3; } message StateTransitionId { diff --git a/crates/p2p/src/conversions/rpc.rs b/crates/p2p/src/conversions/rpc.rs index 0fffde1604..b2f2098819 100644 --- a/crates/p2p/src/conversions/rpc.rs +++ b/crates/p2p/src/conversions/rpc.rs @@ -167,7 +167,11 @@ impl TryFrom for StateTransition { .update .ok_or_else(|| anyhow::anyhow!("Missing state transition update"))?; let update = SubstateUpdate::try_from(update)?; - Ok(Self { id, update }) + Ok(Self { + id, + state_version: value.state_version, + update, + }) } } @@ -176,6 +180,7 @@ impl From for proto::rpc::StateTransition { Self { id: Some(value.id.into()), update: Some(value.update.into()), + state_version: value.state_version, } } } diff --git a/crates/rpc_state_sync/src/state_sync.rs b/crates/rpc_state_sync/src/state_sync.rs index 4b6c8c3ccf..db2d164715 100644 --- a/crates/rpc_state_sync/src/state_sync.rs +++ b/crates/rpc_state_sync/src/state_sync.rs @@ -1,7 +1,10 @@ // Copyright 2023 The Tari Project // SPDX-License-Identifier: BSD-3-Clause -use std::{cmp, collections::HashMap, time::Instant}; +use std::{ + collections::{BTreeMap, HashMap}, + time::Instant, +}; use anyhow::anyhow; use futures::StreamExt; @@ -55,11 +58,11 @@ use tari_template_manager::interface::{TemplateChange, TemplateManagerHandle}; use tari_validator_node_rpc::{ client::{TariValidatorNodeRpcClientFactory, ValidatorNodeClientFactory}, rpc_service::ValidatorNodeRpcClient, + STATE_SYNC_MAX_BATCH_SIZE, }; use crate::{error::RpcStateSyncError, stats::StateSyncStats}; -const BATCH_SIZE: usize = 100; const LOG_TARGET: &str = "tari::ootle::comms_rpc_state_sync"; pub struct RpcStateSyncClientProtocol { @@ -158,22 +161,19 @@ where TConsensusSpec: ConsensusSpec .optional()? .unwrap_or_else(|| StateTransitionId::initial(shard)); - let persisted_version = self + let mut maybe_persisted_state_version = self .state_store .with_read_tx(|tx| tx.state_tree_versions_get_latest(shard))?; if current_epoch == last_state_transition_id.epoch() { info!(target: LOG_TARGET, "πŸ›œAlready up to date. No need to sync."); - return Ok(persisted_version); + return Ok(maybe_persisted_state_version); } - let mut maybe_current_version = persisted_version; - let current_version = maybe_current_version.unwrap_or(0); - info!( target: LOG_TARGET, "πŸ›œSyncing from v{} to state transition {last_state_transition_id}", - current_version + maybe_persisted_state_version.unwrap_or(0) ); self.stats.total_requests += 1; @@ -193,7 +193,7 @@ where TConsensusSpec: ConsensusSpec let msg = match result { Ok(msg) => msg, Err(err) if err.is_not_found() => { - return Ok(maybe_current_version); + return Ok(maybe_persisted_state_version); }, Err(err) => { return Err(err.into()); @@ -205,123 +205,137 @@ where TConsensusSpec: ConsensusSpec "Received empty state transition batch." ))); } + if msg.transitions.len() > STATE_SYNC_MAX_BATCH_SIZE { + return Err(RpcStateSyncError::InvalidResponse(anyhow!( + "Received too many state transitions in a batch: {}. Expected at most {}.", + msg.transitions.len(), + STATE_SYNC_MAX_BATCH_SIZE + ))); + } self.stats.total_transitions += msg.transitions.len() as u64; - tree_changes.reserve_exact(cmp::min(msg.transitions.len(), BATCH_SIZE)); - - self.state_store.with_write_tx(|tx| { - info!( - target: LOG_TARGET, - "πŸ›œ Next state updates batch of size {} from v{}", - msg.transitions.len(), - current_version - ); + // Batch the response into a state_version -> Vec map + let mut transitions = + msg.transitions + .into_iter() + .try_fold(BTreeMap::<_, Vec<_>>::new(), |mut acc, transition| { + let transition = + StateTransition::try_from(transition).map_err(RpcStateSyncError::InvalidResponse)?; + acc.entry(transition.state_version).or_default().push(transition); + Ok::<_, RpcStateSyncError>(acc) + })?; + + loop { + let Some((state_version, transitions_for_state_version)) = transitions.pop_first() else { + info!(target: LOG_TARGET, "πŸ›œ No more state transitions to process for shard {shard}"); + break; + }; + + tree_changes.reserve_exact(transitions_for_state_version.len()); + + self.state_store.with_write_tx(|tx| { + info!( + target: LOG_TARGET, + "πŸ›œ Next state updates batch of size {} from v{}", + transitions_for_state_version.len(), + state_version + ); - let mut store = ShardScopedTreeStoreWriter::new(tx, shard); - - for transition in msg.transitions { - let transition = - StateTransition::try_from(transition).map_err(RpcStateSyncError::InvalidResponse)?; - if transition.id.shard() != shard { - return Err(RpcStateSyncError::InvalidResponse(anyhow!( - "Received state transition for shard {} which is not the expected shard {}.", - transition.id.shard(), - shard - ))); - } + let mut store = ShardScopedTreeStoreWriter::new(tx, shard); - if transition.id.epoch().is_zero() { - return Err(RpcStateSyncError::InvalidResponse(anyhow!( - "Received state transition with epoch 0." - ))); - } - - if transition.id.epoch() >= current_epoch { - return Err(RpcStateSyncError::InvalidResponse(anyhow!( - "Received state transition for epoch {} which is at or ahead of our current epoch {}.", - transition.id.epoch(), - current_epoch - ))); - } + for transition in transitions_for_state_version { + if transition.id.shard() != shard { + return Err(RpcStateSyncError::InvalidResponse(anyhow!( + "Received state transition for shard {} which is not the expected shard {}.", + transition.id.shard(), + shard + ))); + } - let change = match &transition.update { - SubstateUpdate::Create(create) => { - let id = create.substate.as_versioned_substate_id_ref(); - if let Some(template_address) = create.substate.substate_id.as_template() { - match create - .substate - .value - .value() { - Some(value) => { - let template = value.as_template() - .ok_or_else(|| RpcStateSyncError::InvalidResponse( - anyhow!("Validator returned a template address {} but substate value was not a template", id.substate_id()) - ))?; - - info!(target: LOG_TARGET, "πŸ›œ Add template {id}"); - template_changes_mut.push(TemplateChange::Add { - template_address, - author_public_key: template.author, - binary_hash: template.binary_hash.into_array().into(), - epoch: transition.id.epoch(), - }); - } - None => { - // TODO: currently you cannot DOWN a template. If we were to allow deprecations, it would likely be marking the template as deprecated rather than DOWNing it, and not permitting any template (non-component) calls to the template. - // We could still handle this case by requesting the template by address and verifying the template address hash i.e. peers send author and binary. - warn!(target: LOG_TARGET, "❗️ NEVER HAPPEN: Validator sent us a template {} that has no value, indicating it will be DOWNed later. We are not able to sync it", id); - } - }; - } + if transition.id.epoch().is_zero() { + return Err(RpcStateSyncError::InvalidResponse(anyhow!( + "Received state transition with epoch 0." + ))); + } - SubstateTreeChange::Up { - id: id.to_owned(), - value_hash: create.substate.to_value_hash(), - } + if transition.id.epoch() >= current_epoch { + return Err(RpcStateSyncError::InvalidResponse(anyhow!( + "Received state transition for epoch {} which is at or ahead of our current epoch {}.", + transition.id.epoch(), + current_epoch + ))); } - SubstateUpdate::Destroy(destroy) => { - if let Some(template_address) = destroy.substate_id.as_template() { - info!(target: LOG_TARGET, "πŸ›œ Deprecate template {}", template_address); - template_changes_mut.push(TemplateChange::Deprecate { template_address }); - } - SubstateTreeChange::Down { - id: destroy.to_versioned_substate_id() + let change = match &transition.update { + SubstateUpdate::Create(create) => { + let id = create.substate.as_versioned_substate_id_ref(); + if let Some(template_address) = create.substate.substate_id.as_template() { + match create + .substate + .value + .value() { + Some(value) => { + let template = value.as_template() + .ok_or_else(|| RpcStateSyncError::InvalidResponse( + anyhow!("Validator returned a template address {} but substate value was not a template", id.substate_id()) + ))?; + + info!(target: LOG_TARGET, "πŸ›œ Add template {id}"); + template_changes_mut.push(TemplateChange::Add { + template_address, + author_public_key: template.author, + binary_hash: template.binary_hash.into_array().into(), + epoch: transition.id.epoch(), + }); + } + None => { + // TODO: currently you cannot DOWN a template. If we were to allow deprecations, it would likely be marking the template as deprecated rather than DOWNing it, and not permitting any template (non-component) calls to the template. + // We could still handle this case by requesting the template by address and verifying the template address hash i.e. peers send author and binary. + warn!(target: LOG_TARGET, "❗️ NEVER HAPPEN: Validator sent us a template {} that has no value, indicating it will be DOWNed later. We are not able to sync it", id); + } + }; + } + + SubstateTreeChange::Up { + id: id.to_owned(), + value_hash: create.substate.to_value_hash(), + } } - } - }; + SubstateUpdate::Destroy(destroy) => { + if let Some(template_address) = destroy.substate_id.as_template() { + info!(target: LOG_TARGET, "πŸ›œ Deprecate template {}", template_address); + template_changes_mut.push(TemplateChange::Deprecate { template_address }); + } + + SubstateTreeChange::Down { + id: destroy.to_versioned_substate_id() + } + } + }; + info!(target: LOG_TARGET, "πŸ›œ Applying state update (v{}) {}", state_version, transition); + self.commit_update(store.transaction(), checkpoint, checkpoint_block_id, transition)?; + tree_changes.push(change); + } - info!(target: LOG_TARGET, "πŸ›œ Applying state update (v{}) {}", current_version, transition); - self.commit_update(store.transaction(), checkpoint, checkpoint_block_id, transition)?; + info!(target: LOG_TARGET, "πŸ›œ {} state update(s) for v{}", tree_changes.len(), state_version); - tree_changes.push(change); - if tree_changes.len() == BATCH_SIZE { + if !tree_changes.is_empty() { let mut state_tree = SpreadPrefixStateTree::new(&mut store); - let next_version = current_version + 1; - info!(target: LOG_TARGET, "πŸ›œ Committing {} state tree changes v{} to v{}", tree_changes.len(), current_version, next_version); - state_tree.batch_put_substate_changes(maybe_current_version, next_version, tree_changes.drain(..))?; - maybe_current_version = Some(next_version); - store.set_version(next_version)?; + info!(target: LOG_TARGET, "πŸ›œ Committing {} state tree changes batch v{}", tree_changes.len(), state_version); + state_tree.batch_put_substate_changes(maybe_persisted_state_version, state_version, tree_changes.drain(..))?; + maybe_persisted_state_version = Some(state_version); + store.set_version(state_version)?; } - } - - if !tree_changes.is_empty() { - let mut state_tree = SpreadPrefixStateTree::new(&mut store); - let next_version = current_version + 1; - info!(target: LOG_TARGET, "πŸ›œ Committing final {} state tree changes v{} to v{}", tree_changes.len(), current_version, next_version); - state_tree.batch_put_substate_changes(maybe_current_version, next_version, tree_changes.drain(..))?; - maybe_current_version = Some(next_version); - store.set_version(next_version)?; - } - - Ok::<_, RpcStateSyncError>(()) - })?; + + Ok::<_, RpcStateSyncError>(()) + })?; + } } - let local_state_root = self.calculate_state_root_for_shard(shard, maybe_current_version)?; + let local_state_root = self.calculate_state_root_for_shard(shard, maybe_persisted_state_version)?; if local_state_root != checkpoint_state_root { error!( target: LOG_TARGET, @@ -337,9 +351,9 @@ where TConsensusSpec: ConsensusSpec }); } - info!(target: LOG_TARGET, "πŸ›œ Synced state for {shard} to v{} with root {local_state_root}", maybe_current_version.unwrap_or(0)); + info!(target: LOG_TARGET, "πŸ›œ Synced state for {shard} to v{} with root {local_state_root}", maybe_persisted_state_version.unwrap_or(0)); - Ok(maybe_current_version) + Ok(maybe_persisted_state_version) } fn calculate_state_root_for_shard( diff --git a/crates/state_store_rocksdb/src/column_families/state_transition.rs b/crates/state_store_rocksdb/src/column_families/state_transition.rs index 58ae0ae678..99abffa6cc 100644 --- a/crates/state_store_rocksdb/src/column_families/state_transition.rs +++ b/crates/state_store_rocksdb/src/column_families/state_transition.rs @@ -23,6 +23,7 @@ use serde::{Deserialize, Serialize}; use tari_ootle_common_types::{shard::Shard, SubstateAddress}; use tari_ootle_storage::consensus_models::StateTransitionId; +use tari_state_tree::Version; use crate::{ codecs::{DefaultCodec, EpochCodec, NumberCodec, ShardCodec, StateTransitionIdCodec}, @@ -33,6 +34,7 @@ use crate::{ pub struct StateTransitionModelData { pub substate_address: SubstateAddress, pub transition: StateTransitionType, + pub state_version: Version, } #[derive(Debug, Clone, Copy, Serialize, Deserialize)] diff --git a/crates/state_store_rocksdb/src/reader.rs b/crates/state_store_rocksdb/src/reader.rs index b90cb26cf4..d54378909d 100644 --- a/crates/state_store_rocksdb/src/reader.rs +++ b/crates/state_store_rocksdb/src/reader.rs @@ -1646,7 +1646,11 @@ impl<'tx, TAddr: NodeAddressable + Serialize + DeserializeOwned + 'tx> StateStor }), }; - transitions.push(StateTransition { id: key, update }); + transitions.push(StateTransition { + id: key, + state_version: value.state_version, + update, + }); if transitions.len() == n { break; } diff --git a/crates/state_store_rocksdb/src/writer.rs b/crates/state_store_rocksdb/src/writer.rs index 7d19faec0a..e2e2da8a1f 100644 --- a/crates/state_store_rocksdb/src/writer.rs +++ b/crates/state_store_rocksdb/src/writer.rs @@ -1216,6 +1216,12 @@ impl<'tx, TAddr: NodeAddressable + 'tx> StateStoreWriteTransaction for RocksDbSt OPERATION, )?; + let shard_state_version = db + .cf(StateTreeShardVersionCf)? + .get(&substate.created_by_shard, OPERATION) + .optional()? + .unwrap_or_default(); + let seq_index = db.cf(state_transition::ShardSeqIndex)?; let seq = seq_index.get(&substate.created_by_shard, OPERATION).optional()?; let next_seq = seq.map(|s| s + 1).unwrap_or(1); @@ -1223,6 +1229,7 @@ impl<'tx, TAddr: NodeAddressable + 'tx> StateStoreWriteTransaction for RocksDbSt let id = StateTransitionId::new(substate.created_at_epoch, substate.created_by_shard, next_seq); let transition = StateTransitionModelData { substate_address: address, + state_version: shard_state_version, transition: StateTransitionType::Up, }; @@ -1269,8 +1276,16 @@ impl<'tx, TAddr: NodeAddressable + 'tx> StateStoreWriteTransaction for RocksDbSt let next_seq = seq.map(|s| s + 1).unwrap_or(1); let transitions_cf = db.cf(StateTransitionCf)?; + + let shard_state_version = db + .cf(StateTreeShardVersionCf)? + .get(&shard, OPERATION) + .optional()? + .unwrap_or_default(); + let data = StateTransitionModelData { substate_address: address, + state_version: shard_state_version, transition: StateTransitionType::Down, }; let id = StateTransitionId::new(epoch, shard, next_seq); diff --git a/crates/storage/src/consensus_models/state_transition.rs b/crates/storage/src/consensus_models/state_transition.rs index 1906a9f01c..77c4addb15 100644 --- a/crates/storage/src/consensus_models/state_transition.rs +++ b/crates/storage/src/consensus_models/state_transition.rs @@ -9,13 +9,14 @@ use std::{ use serde::{Deserialize, Serialize}; use tari_ootle_common_types::{shard::Shard, Epoch}; -use tari_state_tree::SubstateTreeChange; +use tari_state_tree::{SubstateTreeChange, Version}; use crate::{consensus_models::SubstateUpdate, StateStoreReadTransaction, StorageError}; #[derive(Debug, Clone, Serialize, Deserialize)] pub struct StateTransition { pub id: StateTransitionId, + pub state_version: Version, pub update: SubstateUpdate, } diff --git a/crates/validator_node_rpc/src/lib.rs b/crates/validator_node_rpc/src/lib.rs index 2665e173a9..1cf34b1f5f 100644 --- a/crates/validator_node_rpc/src/lib.rs +++ b/crates/validator_node_rpc/src/lib.rs @@ -24,3 +24,5 @@ pub mod client; mod error; pub mod rpc_service; pub use error::ValidatorNodeRpcClientError; + +pub const STATE_SYNC_MAX_BATCH_SIZE: usize = 100;