fix(consensus): dont send dummy blocks on sync (#886)

Description
---
Excludes dummy blocks from catch-up sync
Excludes dummy blocks from RPC sync
Rename `tari_comms_rpc_state_sync` to `tari_rpc_state_sync`
Removed commented-out and dead code

Motivation and Context
---
The is_dummy flag is not sent on catch-up, which leads to a "missing
proposal signature" error, since summy blocks cannot be signed. Dummy
blocks are calculated as needed by the proposal handler so do not need
to be sent.

When testing manually this works, however, we sometimes run into #887 

How Has This Been Tested?
---
Manually, synced chain with dummy blocks and checked that dummy blocks
created

What process can a PR reviewer use to test or verify this change?
---
4 VN committee, shut down one node and delete its data, wait for leader
failures (to create some dummy blocks), restart node, and check that it
syncs

Breaking Changes
---

- [x] None
- [ ] Requires data directory to be deleted
- [ ] Other - Please specify

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

@@ -127,7 +127,7 @@ where TConsensusSpec: ConsensusSpec
         );
 
         if block.height() < current_height {
-            debug!(
+            info!(
                 target: LOG_TARGET,
                 "🔥 Block {} is lower than current height {}. Ignoring.",
                 block,
@@ -302,7 +302,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 {