fix(consensus): dont send dummy blocks on catch up

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/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,

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 {

dan_layer/consensus/src/hotstuff/worker.rs

@@ -29,7 +29,7 @@ use crate::{
         on_next_sync_view::OnNextSyncViewHandler,
         on_propose::OnPropose,
         on_receive_foreign_proposal::OnReceiveForeignProposalHandler,
-        on_receive_local_proposal::OnReceiveProposalHandler,
+        on_receive_local_proposal::OnReceiveLocalProposalHandler,
         on_receive_new_view::OnReceiveNewViewHandler,
         on_receive_request_missing_transactions::OnReceiveRequestMissingTransactions,
         on_receive_vote::OnReceiveVoteHandler,
@@ -52,7 +52,7 @@ pub struct HotstuffWorker<TConsensusSpec: ConsensusSpec> {
     inbound_message_worker: OnInboundMessage<TConsensusSpec>,
 
     on_next_sync_view: OnNextSyncViewHandler<TConsensusSpec>,
-    on_receive_local_proposal: OnReceiveProposalHandler<TConsensusSpec>,
+    on_receive_local_proposal: OnReceiveLocalProposalHandler<TConsensusSpec>,
     on_receive_foreign_proposal: OnReceiveForeignProposalHandler<TConsensusSpec>,
     on_receive_vote: OnReceiveVoteHandler<TConsensusSpec>,
     on_receive_new_view: OnReceiveNewViewHandler<TConsensusSpec>,
@@ -120,7 +120,7 @@ impl<TConsensusSpec: ConsensusSpec> HotstuffWorker<TConsensusSpec> {
                 leader_strategy.clone(),
                 epoch_manager.clone(),
             ),
-            on_receive_local_proposal: OnReceiveProposalHandler::new(
+            on_receive_local_proposal: OnReceiveLocalProposalHandler::new(
                 validator_addr,
                 state_store.clone(),
                 epoch_manager.clone(),

dan_layer/state_store_sqlite/src/reader.rs

@@ -609,6 +609,7 @@ impl<TAddr: NodeAddressable + Serialize + DeserializeOwned> StateStoreReadTransa
         &mut self,
         start_block_id_exclusive: &BlockId,
         end_block_id_inclusive: &BlockId,
+        include_dummy_blocks: bool,
     ) -> Result<Vec<Block>, StorageError> {
         use crate::schema::{blocks, quorum_certificates};
 
@@ -620,10 +621,17 @@ impl<TAddr: NodeAddressable + Serialize + DeserializeOwned> StateStoreReadTransa
         // Exclude start block
         block_ids.pop();
 
-        let results = blocks::table
+        let mut query = blocks::table
             .left_join(quorum_certificates::table.on(blocks::qc_id.eq(quorum_certificates::qc_id)))
             .select((blocks::all_columns, quorum_certificates::all_columns.nullable()))
             .filter(blocks::block_id.eq_any(block_ids))
+            .boxed();
+
+        if include_dummy_blocks {
+            query = query.filter(blocks::is_dummy.eq(false)).boxed();
+        }
+
+        let results = query
             .order_by(blocks::height.asc())
             .get_results::<(sql_models::Block, Option<sql_models::QuorumCertificate>)>(self.connection())
             .map_err(|e| SqliteStorageError::DieselError {

dan_layer/storage/src/consensus_models/block.rs

@@ -358,8 +358,9 @@ impl Block {
         tx: &mut TTx,
         start_block_id_exclusive: &BlockId,
         end_block_id_inclusive: &BlockId,
+        include_dummy_blocks: bool,
     ) -> Result<Vec<Self>, StorageError> {
-        tx.blocks_get_all_between(start_block_id_exclusive, end_block_id_inclusive)
+        tx.blocks_get_all_between(start_block_id_exclusive, end_block_id_inclusive, include_dummy_blocks)
     }
 
     pub fn exists<TTx: StateStoreReadTransaction + ?Sized>(&self, tx: &mut TTx) -> Result<bool, StorageError> {

dan_layer/storage/src/state_store/mod.rs

@@ -121,6 +121,7 @@ pub trait StateStoreReadTransaction {
         &mut self,
         start_block_id_exclusive: &BlockId,
         end_block_id_inclusive: &BlockId,
+        include_dummy_blocks: bool,
     ) -> Result<Vec<Block>, StorageError>;
     fn blocks_exists(&mut self, block_id: &BlockId) -> Result<bool, StorageError>;
     fn blocks_is_ancestor(&mut self, descendant: &BlockId, ancestor: &BlockId) -> Result<bool, StorageError>;

networking/libp2p-peersync/src/error.rs

@@ -30,8 +30,6 @@ pub enum Error {
     InvalidMessage { peer_id: PeerId, details: String },
     #[error("Failed to decode multiaddr: {0}")]
     DecodeMultiaddr(#[from] multiaddr::Error),
-    #[error("ProtoBuf peer record has no signature")]
-    ProtoBufMissingSignature,
 
     #[error("Invalid signed peer receord from peer `{peer_id}`: {details}")]
     InvalidSignedPeer { peer_id: PeerId, details: String },