fix(state_sync)!: sync state version

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<TStateStore: StateStore + Clone + Send + Sync + 'static> ValidatorNodeRpcSe
                 sender,
                 last_state_transition_for_chain,
                 end_epoch,
+                STATE_SYNC_MAX_BATCH_SIZE,
             )
             .run(),
         );

applications/tari_validator_node/src/p2p/rpc/state_sync_task.rs

@@ -14,15 +14,14 @@ use tokio::sync::mpsc;
 
 const LOG_TARGET: &str = "tari::ootle::rpc::sync_task";
 
-const BATCH_SIZE: usize = 100;
-
 type UpdateBuffer = Vec<StateTransition>;
 
 pub struct StateSyncTask<TStateStore: StateStore> {
     store: TStateStore,
     sender: mpsc::Sender<Result<SyncStateResponse, RpcStatus>>,
     start_state_transition_id: StateTransitionId,
     current_epoch: Epoch,
+    batch_size: usize,
 }
 
 impl<TStateStore: StateStore> StateSyncTask<TStateStore> {
@@ -31,17 +30,19 @@ impl<TStateStore: StateStore> StateSyncTask<TStateStore> {
         sender: mpsc::Sender<Result<SyncStateResponse, RpcStatus>>,
         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<TStateStore: StateStore> StateSyncTask<TStateStore> {
     ) -> Result<Option<StateTransitionId>, 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();
 

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 {

crates/p2p/src/conversions/rpc.rs

@@ -167,7 +167,11 @@ impl TryFrom<proto::rpc::StateTransition> 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<StateTransition> for proto::rpc::StateTransition {
         Self {
             id: Some(value.id.into()),
             update: Some(value.update.into()),
+            state_version: value.state_version,
         }
     }
 }