fix(consensus): dont send dummy blocks on sync

Cargo.toml

@@ -19,12 +19,12 @@ members = [
     "clients/base_node_client",
     "clients/validator_node_client",
     "clients/wallet_daemon_client",
-    "dan_layer/comms_rpc_state_sync",
     "dan_layer/consensus",
     "dan_layer/consensus_tests",
     "dan_layer/epoch_manager",
     "dan_layer/indexer_lib",
     "dan_layer/p2p",
+    "dan_layer/rpc_state_sync",
     "dan_layer/state_store_sqlite",
     "dan_layer/storage_lmdb",
     "dan_layer/storage_sqlite",
@@ -64,9 +64,9 @@ libp2p-substream = { path = "networking/libp2p-substream" }
 proto_builder = { path = "networking/proto_builder" }
 sqlite_message_logger = { path = "networking/sqlite_message_logger" }
 tari_base_node_client = { path = "clients/base_node_client" }
-# avoid including default features so each crate can choose which ones to import
+# avoid including default features so each crate can choose which ones to import
 tari_bor = { version = "0.3.0", path = "dan_layer/tari_bor", default-features = false }
-tari_comms_rpc_state_sync = { path = "dan_layer/comms_rpc_state_sync" }
+tari_rpc_state_sync = { path = "dan_layer/rpc_state_sync" }
 tari_consensus = { path = "dan_layer/consensus" }
 tari_dan_app_utilities = { path = "applications/tari_dan_app_utilities" }
 tari_dan_common_types = { path = "dan_layer/common_types" }

applications/tari_validator_node/Cargo.toml

@@ -12,7 +12,7 @@ minotari_app_grpc = { workspace = true }
 minotari_app_utilities = { workspace = true }
 minotari_wallet_grpc_client = { workspace = true }
 tari_common = { workspace = true }
-tari_common_types = {workspace = true }
+tari_common_types = { workspace = true }
 tari_core = { workspace = true, default-features = false, features = ["transactions"] }
 tari_crypto = { workspace = true }
 tari_validator_node_rpc = { workspace = true }
@@ -30,7 +30,7 @@ tari_validator_node_client = { workspace = true }
 tari_base_node_client = { workspace = true }
 tari_epoch_manager = { workspace = true, features = ["base_layer"] }
 tari_indexer_lib = { workspace = true }
-tari_comms_rpc_state_sync = {workspace = true }
+tari_rpc_state_sync = { workspace = true }
 tari_bor = { workspace = true, default-features = true }
 tari_consensus = { workspace = true }
 tari_state_store_sqlite = { workspace = true }
@@ -72,7 +72,7 @@ tokio = { workspace = true, features = [
     "time",
     "sync",
     "rt-multi-thread",
-]}
+] }
 tonic = { workspace = true }
 tower-http = { workspace = true, features = ["default", "cors"] }
 

applications/tari_validator_node/src/consensus/mod.rs

@@ -1,7 +1,6 @@
 //    Copyright 2023 The Tari Project
 //    SPDX-License-Identifier: BSD-3-Clause
 
-use tari_comms_rpc_state_sync::CommsRpcStateSyncManager;
 use tari_consensus::{
     hotstuff::{ConsensusWorker, ConsensusWorkerContext, HotstuffWorker},
     messages::HotstuffMessage,
@@ -39,6 +38,7 @@ pub use handle::*;
 use sqlite_message_logger::SqliteMessageLogger;
 use tari_dan_app_utilities::keypair::RistrettoKeypair;
 use tari_dan_common_types::PeerAddress;
+use tari_rpc_state_sync::RpcStateSyncManager;
 
 use crate::p2p::services::message_dispatcher::OutboundMessaging;
 
@@ -90,7 +90,7 @@ pub async fn spawn(
     let context = ConsensusWorkerContext {
         epoch_manager: epoch_manager.clone(),
         hotstuff: hotstuff_worker,
-        state_sync: CommsRpcStateSyncManager::new(epoch_manager, store, client_factory),
+        state_sync: RpcStateSyncManager::new(epoch_manager, store, leader_strategy, client_factory),
         tx_current_state,
     };
 

applications/tari_validator_node/src/consensus/spec.rs

@@ -1,10 +1,10 @@
 //    Copyright 2023 The Tari Project
 //    SPDX-License-Identifier: BSD-3-Clause
 
-use tari_comms_rpc_state_sync::CommsRpcStateSyncManager;
 use tari_consensus::traits::ConsensusSpec;
 use tari_dan_common_types::PeerAddress;
 use tari_epoch_manager::base_layer::EpochManagerHandle;
+use tari_rpc_state_sync::RpcStateSyncManager;
 use tari_state_store_sqlite::SqliteStateStore;
 
 use crate::consensus::{
@@ -23,5 +23,5 @@ impl ConsensusSpec for TariConsensusSpec {
     type SignatureService = TariSignatureService;
     type StateManager = TariStateManager;
     type StateStore = SqliteStateStore<Self::Addr>;
-    type SyncManager = CommsRpcStateSyncManager<Self::EpochManager, Self::StateStore>;
+    type SyncManager = RpcStateSyncManager<Self>;
 }

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

@@ -109,13 +109,19 @@ impl<TStateStore: StateStore> BlockSyncTask<TStateStore> {
     fn fetch_next_batch(&self, buffer: &mut BlockBuffer, current_block_id: &BlockId) -> Result<BlockId, StorageError> {
         self.store.with_read_tx(|tx| {
             let mut current_block_id = *current_block_id;
+            let mut last_block_id = current_block_id;
             loop {
                 let children = tx.blocks_get_all_by_parent(&current_block_id)?;
                 let Some(child) = children.into_iter().find(|b| b.is_committed()) else {
                     break;
                 };
 
                 current_block_id = *child.id();
+                if child.is_dummy() {
+                    continue;
+                }
+
+                last_block_id = current_block_id;
                 let all_qcs = child
                     .commands()
                     .iter()
@@ -130,15 +136,15 @@ impl<TStateStore: StateStore> BlockSyncTask<TStateStore> {
                     break;
                 }
             }
-            Ok::<_, StorageError>(current_block_id)
+            Ok::<_, StorageError>(last_block_id)
         })
     }
 
     fn fetch_last_blocks(&self, buffer: &mut BlockBuffer, current_block_id: &BlockId) -> Result<(), StorageError> {
         self.store.with_read_tx(|tx| {
             // TODO: if there are any transactions this will break the syncing node.
             let leaf_block = LeafBlock::get(tx)?;
-            let blocks = Block::get_all_blocks_between(tx, current_block_id, leaf_block.block_id())?;
+            let blocks = Block::get_all_blocks_between(tx, current_block_id, leaf_block.block_id(), false)?;
             for block in blocks {
                 debug!(
                     target: LOG_TARGET,

dan_layer/consensus/src/hotstuff/error.rs

@@ -148,7 +148,7 @@ pub enum ProposalValidationError {
     },
     #[error("Proposed block {block_id} {height} already has been processed")]
     BlockAlreadyProcessed { block_id: BlockId, height: NodeHeight },
-    #[error("Proposed block {block_id} {height} doesn't have signature")]
+    #[error("Proposed block {block_id} {height} doesn't have a signature")]
     MissingSignature { block_id: BlockId, height: NodeHeight },
     #[error("Proposed block {block_id} {height} has invalid signature")]
     InvalidSignature { block_id: BlockId, height: NodeHeight },

dan_layer/consensus/src/hotstuff/mod.rs

@@ -26,6 +26,7 @@ mod state_machine;
 mod vote_receiver;
 mod worker;
 
+pub use common::*;
 pub use error::*;
 pub use event::*;
 pub use state_machine::*;

dan_layer/consensus/src/hotstuff/on_inbound_message.rs

@@ -145,7 +145,7 @@ where TConsensusSpec: ConsensusSpec
         );
 
         if block.height() < current_height {
-            debug!(
+            info!(
                 target: LOG_TARGET,
                 "🔥 Block {} is lower than current height {}. Ignoring.",
                 block,
@@ -319,7 +319,11 @@ impl<TAddr: NodeAddressable> MessageBuffer<TAddr> {
                 },
                 // Buffer message for future height
                 Some(h) if h > current_height => {
-                    debug!(target: LOG_TARGET, "Message {} is for future block {}. Current height {}", msg, h, current_height);
+                    if msg.proposal().is_some() {
+                        info!(target: LOG_TARGET, "Proposal {} is for future block {}. Current height {}", msg, h, current_height);
+                    } else {
+                        debug!(target: LOG_TARGET, "Message {} is for future height {}. Current height {}", msg, h, current_height);
+                    }
                     self.push_to_buffer(h, from, msg);
                     continue;
                 },

dan_layer/consensus/src/hotstuff/on_receive_local_proposal.rs

@@ -36,15 +36,15 @@ use crate::{
 
 const LOG_TARGET: &str = "tari::dan::consensus::hotstuff::on_receive_local_proposal";
 
-pub struct OnReceiveProposalHandler<TConsensusSpec: ConsensusSpec> {
+pub struct OnReceiveLocalProposalHandler<TConsensusSpec: ConsensusSpec> {
     store: TConsensusSpec::StateStore,
     epoch_manager: TConsensusSpec::EpochManager,
     leader_strategy: TConsensusSpec::LeaderStrategy,
     pacemaker: PaceMakerHandle,
     on_ready_to_vote_on_local_block: OnReadyToVoteOnLocalBlock<TConsensusSpec>,
 }
 
-impl<TConsensusSpec: ConsensusSpec> OnReceiveProposalHandler<TConsensusSpec> {
+impl<TConsensusSpec: ConsensusSpec> OnReceiveLocalProposalHandler<TConsensusSpec> {
     pub fn new(
         validator_addr: TConsensusSpec::Addr,
         store: TConsensusSpec::StateStore,
@@ -137,6 +137,13 @@ impl<TConsensusSpec: ConsensusSpec> OnReceiveProposalHandler<TConsensusSpec> {
         self.store.with_write_tx(|tx| {
             match self.validate_local_proposed_block(tx, block, &local_committee, &local_committee_shard) {
                 Ok(validated) => Ok(Some(validated)),
+                // Propagate this error out as sync is needed in the case where we have a valid QC but do not know the
+                // block
+                Err(
+                    err @ HotStuffError::ProposalValidationError(ProposalValidationError::JustifyBlockNotFound {
+                        ..
+                    }),
+                ) => Err(err),
                 // Validation errors should not cause a FAILURE state transition
                 Err(HotStuffError::ProposalValidationError(err)) => {
                     warn!(target: LOG_TARGET, "❌ Block failed validation: {}", err);

dan_layer/consensus/src/hotstuff/on_sync_request.rs

@@ -48,7 +48,7 @@ impl<TConsensusSpec: ConsensusSpec> OnSyncRequest<TConsensusSpec> {
                     msg.high_qc,
                     last_voted
                 );
-                let blocks = Block::get_all_blocks_between(tx, msg.high_qc.block_id(), last_voted.block_id())?;
+                let blocks = Block::get_all_blocks_between(tx, msg.high_qc.block_id(), last_voted.block_id(), false)?;
 
                 debug!(
                     target: LOG_TARGET,
@@ -58,25 +58,6 @@ impl<TConsensusSpec: ConsensusSpec> OnSyncRequest<TConsensusSpec> {
                 );
 
                 Ok::<_, HotStuffError>(blocks)
-
-                // let mut full_blocks = Vec::with_capacity(blocks.len());
-                // for block in blocks {
-                //     let all_qcs = block
-                //         .commands()
-                //         .iter()
-                //         .flat_map(|cmd| cmd.evidence().qc_ids_iter())
-                //         .collect::<HashSet<_>>();
-                //     let qcs = QuorumCertificate::get_all(tx, all_qcs)?;
-                //     let transactions = block.get_transactions(tx)?;
-                //
-                //     full_blocks.push(FullBlock {
-                //         block,
-                //         qcs,
-                //         transactions: transactions.into_iter().map(|t| t.into_transaction()).collect(),
-                //     });
-                // }
-                //
-                // Ok::<_, HotStuffError>(full_blocks)
             });
 
             let blocks = match result {