refactor(state_sync)!: sync state transitions by (shard,state_version)

Cargo.toml

@@ -165,6 +165,7 @@ blake2 = "0.10.6"
 borsh = { version = "1.5", default-features = false }
 bytes = "1.10.0"
 bnum = "0.13.0"
+bounded-vec = "0.9.0"
 cacache = "12.0.0"
 cargo_toml = "0.20.5"
 ciborium = { version = "0.2.2", default-features = false }

applications/tari_indexer/src/block_data.rs

@@ -1,10 +1,10 @@
 //   Copyright 2025 The Tari Project
 //   SPDX-License-Identifier: BSD-3-Clause
 
-use tari_ootle_storage::consensus_models::{Block, SubstateUpdate};
+use tari_ootle_storage::consensus_models::{Block, SubstateUpdateProof};
 
 #[derive(Clone, Debug)]
 pub struct BlockData {
     pub block: Block,
-    pub diff: Vec<SubstateUpdate>,
+    pub diff: Vec<SubstateUpdateProof>,
 }

applications/tari_indexer/src/event_scanner.rs

@@ -33,7 +33,7 @@ use tari_engine_types::{
 use tari_epoch_manager::{service::EpochManagerHandle, EpochManagerReader};
 use tari_ootle_common_types::{committee::Committee, Epoch, PeerAddress, ShardGroup};
 use tari_ootle_p2p::{proto, proto::rpc::SyncBlocksRequest};
-use tari_ootle_storage::consensus_models::{Block, SubstateUpdate};
+use tari_ootle_storage::consensus_models::{Block, SubstateUpdateProof};
 use tari_template_lib::types::{EntityId, TemplateAddress};
 use tari_template_manager::interface::{TemplateChange, TemplateManagerHandle};
 use tari_validator_node_rpc::client::{TariValidatorNodeRpcClientFactory, ValidatorNodeClientFactory};
@@ -309,7 +309,7 @@ impl EventScanner {
         Ok(())
     }
 
-    fn store_substates_in_db(&self, updates: &[SubstateUpdate], timestamp: u64) -> Result<(), anyhow::Error> {
+    fn store_substates_in_db(&self, updates: &[SubstateUpdateProof], timestamp: u64) -> Result<(), anyhow::Error> {
         let mut tx = self.substate_store.create_write_tx()?;
         // store/update up substates if any
         for create in updates.iter().filter_map(|up| up.as_create()) {
@@ -527,7 +527,7 @@ impl EventScanner {
                 let update = msg
                     .into_substate_update()
                     .ok_or_else(|| anyhow::anyhow!("Expected a substate"))?;
-                let update = SubstateUpdate::try_from(update)?;
+                let update = SubstateUpdateProof::try_from(update)?;
                 diff.push(update);
             }
 

applications/tari_validator_node/Cargo.toml

@@ -46,6 +46,7 @@ tari_networking = { workspace = true }
 tari_rpc_framework = { workspace = true }
 tari_template_builtin = { workspace = true }
 tari_swarm = { workspace = true }
+tari_state_tree = { workspace = true }
 tari_sidechain = { workspace = true }
 
 sqlite_message_logger = { workspace = true }

applications/tari_validator_node/src/bootstrap.rs

@@ -219,16 +219,9 @@ pub async fn spawn_services(
 
     info!(target: LOG_TARGET, "State store initializing");
 
-    let sidechain_id = config
-        .validator_node
-        .validator_node_sidechain_id
-        .as_ref()
-        .map(|pk| pk.to_byte_type());
-
     let state_store = ValidatorNodeStateStore::open(&config.validator_node.state_db_path, DatabaseOptions::default())?;
 
-    state_store
-        .with_write_tx(|tx| bootstrap_state(tx, config.network, consensus_constants.num_preshards, sidechain_id))?;
+    state_store.with_write_tx(|tx| bootstrap_state(tx, config.network, consensus_constants.num_preshards))?;
 
     info!(target: LOG_TARGET, "Epoch manager initializing");
     let epoch_manager_config = EpochManagerConfig {
@@ -335,6 +328,12 @@ pub async fn spawn_services(
     #[cfg(not(feature = "metrics"))]
     let metrics = NoopHooks;
 
+    let sidechain_id = config
+        .validator_node
+        .validator_node_sidechain_id
+        .as_ref()
+        .map(|pk| pk.to_byte_type());
+
     let signing_service = consensus::TariSignatureService::new(keypair.clone());
     let (consensus_join_handle, consensus_handle) = consensus::spawn(
         config.network,

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

@@ -12,7 +12,7 @@ use tari_ootle_p2p::{
     proto::rpc::{sync_blocks_response::SyncData, QuorumCertificates, SyncBlocksResponse},
 };
 use tari_ootle_storage::{
-    consensus_models::{Block, SubstateCreatedProof, SubstateUpdate, TransactionRecord},
+    consensus_models::{Block, SubstateCreatedProof, SubstateUpdateProof, TransactionRecord},
     StateStore,
     StateStoreReadTransaction,
     StorageError,
@@ -27,7 +27,7 @@ const BLOCK_BUFFER_SIZE: usize = 15;
 struct BlockData {
     block: Block,
     qcs: Vec<ProposalCertificate>,
-    substates: Vec<SubstateUpdate>,
+    substates: Vec<SubstateUpdateProof>,
     transactions: Vec<TransactionRecord>,
     transaction_receipts: Vec<SubstateCreatedProof>,
 }

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

@@ -26,7 +26,16 @@ use log::*;
 use tari_bor::encode;
 use tari_consensus_types::BlockId;
 use tari_epoch_manager::{service::EpochManagerHandle, EpochManagerReader};
-use tari_ootle_common_types::{optional::Optional, shard::Shard, Epoch, NodeHeight, PeerAddress, SubstateRequirement};
+use tari_ootle_common_types::{
+    displayable::Displayable,
+    optional::Optional,
+    shard::Shard,
+    Epoch,
+    NodeHeight,
+    NumPreshards,
+    PeerAddress,
+    SubstateRequirement,
+};
 use tari_ootle_p2p::{
     proto,
     proto::rpc::{
@@ -47,7 +56,7 @@ use tari_ootle_p2p::{
     },
 };
 use tari_ootle_storage::{
-    consensus_models::{Block, EpochCheckpoint, StateTransitionId, SubstateRecord, TransactionRecord},
+    consensus_models::{Block, EpochCheckpoint, SubstateRecord, TransactionRecord},
     StateStore,
 };
 use tari_rpc_framework::{Request, Response, RpcStatus, Streaming};
@@ -171,27 +180,14 @@ impl<TStateStore: StateStore + Clone + Send + Sync + 'static> ValidatorNodeRpcSe
             }));
         };
 
-        let created_qc = substate
-            .get_created_proposal_certificate(&tx)
-            // TODO: We may not have this... We dont sync PCs. Hmm...
-            // This does not actually prove this substate was committed anyhow.
-            .optional()
-            .map_err(RpcStatus::log_internal_error(LOG_TARGET))?;
-
-        let resp = if substate.is_destroyed() {
-            let destroyed_qc = substate
-                .get_destroyed_proposal_certificate(&tx)
-                .map_err(RpcStatus::log_internal_error(LOG_TARGET))?;
+        let resp = if let Some(destroyed) = substate.destroyed() {
             GetSubstateResponse {
                 status: SubstateStatus::Down as i32,
                 address: substate.substate_id().to_bytes(),
+                substate: vec![],
                 version: substate.version(),
-                quorum_certificates: created_qc
-                    .into_iter()
-                    .chain(destroyed_qc)
-                    .map(|qc| (&qc).into())
-                    .collect(),
-                ..Default::default()
+                created_at_state_version: substate.created().at_state_version,
+                destroyed_at_state_version: destroyed.at_state_version,
             }
         } else {
             GetSubstateResponse {
@@ -202,7 +198,8 @@ impl<TStateStore: StateStore + Clone + Send + Sync + 'static> ValidatorNodeRpcSe
                     .substate_value()
                     .map(|v| v.to_bytes())
                     .ok_or_else(|| RpcStatus::general("NEVER HAPPEN: UP substate has no value"))?,
-                quorum_certificates: created_qc.iter().map(Into::into).collect(),
+                created_at_state_version: substate.created().at_state_version,
+                destroyed_at_state_version: Default::default(),
             }
         };
 
@@ -380,20 +377,31 @@ impl<TStateStore: StateStore + Clone + Send + Sync + 'static> ValidatorNodeRpcSe
 
         let (sender, receiver) = mpsc::channel(10);
 
-        let start_epoch = Epoch(req.start_epoch);
-        let start_shard = Shard::from(req.start_shard);
-        let last_state_transition_for_chain = StateTransitionId::new(start_epoch, start_shard, req.start_seq);
+        let shard = Shard::from_u32(req.shard);
+        if shard > NumPreshards::MAX_SHARD {
+            return Err(RpcStatus::bad_request(format!(
+                "Shard {} out of range. Maximum shard is {}",
+                shard,
+                NumPreshards::MAX_SHARD
+            )));
+        }
 
-        let end_epoch = Epoch(req.current_epoch);
-        info!(target: LOG_TARGET, "🌍peer initiated sync with this node ({}, {}, seq={}) to {}", start_epoch, start_shard, req.start_seq, end_epoch);
+        let end_epoch = Some(req.until_epoch).filter(|e| *e > 0).map(Epoch::from);
+        info!(target: LOG_TARGET, "🌍peer initiated sync with this node (start: v{}, {}) to {}", req.start_state_version, shard, end_epoch.display());
+        if req.start_state_version == 0 {
+            return Err(RpcStatus::bad_request("start_state_version must be greater than 0"));
+        }
 
         task::spawn(
             StateSyncTask::new(
                 self.state_store.clone(),
                 sender,
-                last_state_transition_for_chain,
+                shard,
+                req.start_state_version,
                 end_epoch,
-                STATE_SYNC_MAX_BATCH_SIZE,
+                STATE_SYNC_MAX_BATCH_SIZE
+                    .try_into()
+                    .expect("STATE_SYNC_MAX_BATCH_SIZE is not zero"),
             )
             .run(),
         );