Merge branch 'development' into consensus-sync-no-dummy-blocks

applications/tari_validator_node/src/json_rpc/handlers.rs

@@ -37,6 +37,7 @@ use tari_dan_storage::{
     consensus_models::{Block, ExecutedTransaction, LeafBlock, QuorumDecision, SubstateRecord, TransactionRecord},
     Ordering,
     StateStore,
+    StateStoreReadTransaction,
 };
 use tari_epoch_manager::{base_layer::EpochManagerHandle, EpochManagerReader};
 use tari_networking::{is_supported_multiaddr, NetworkingHandle, NetworkingService};
@@ -292,6 +293,16 @@ impl JsonRpcHandlers {
         Ok(JsonRpcResponse::success(answer_id, res))
     }
 
+    pub async fn get_tx_pool(&self, value: JsonRpcExtractor) -> JrpcResult {
+        let answer_id = value.get_answer_id();
+        let tx_pool = self
+            .state_store
+            .with_read_tx(|tx| tx.transaction_pool_get_all())
+            .map_err(internal_error(answer_id))?;
+        let res = json!({ "tx_pool": tx_pool });
+        Ok(JsonRpcResponse::success(answer_id, res))
+    }
+
     pub async fn get_transaction_result(&self, value: JsonRpcExtractor) -> JrpcResult {
         let answer_id = value.get_answer_id();
         let request: GetTransactionResultRequest = value.parse_params()?;

applications/tari_validator_node/src/json_rpc/server.rs

@@ -68,6 +68,7 @@ async fn handler(Extension(handlers): Extension<Arc<JsonRpcHandlers>>, value: Js
         "get_substates_created_by_transaction" => handlers.get_substates_created_by_transaction(value).await,
         "get_substates_destroyed_by_transaction" => handlers.get_substates_destroyed_by_transaction(value).await,
         "list_blocks" => handlers.list_blocks(value).await,
+        "get_tx_pool" => handlers.get_tx_pool(value).await,
         // Blocks
         "get_block" => handlers.get_block(value).await,
         "get_blocks_count" => handlers.get_blocks_count(value).await,

dan_layer/consensus/src/hotstuff/on_inbound_message.rs

@@ -35,11 +35,11 @@ pub struct OnInboundMessage<TConsensusSpec: ConsensusSpec> {
     leader_strategy: TConsensusSpec::LeaderStrategy,
     pacemaker: PaceMakerHandle,
     vote_signing_service: TConsensusSpec::SignatureService,
-    rx_hotstuff_message: mpsc::Receiver<(TConsensusSpec::Addr, HotstuffMessage)>,
+    pub rx_hotstuff_message: mpsc::Receiver<(TConsensusSpec::Addr, HotstuffMessage)>,
     tx_outbound_message: mpsc::Sender<(TConsensusSpec::Addr, HotstuffMessage)>,
     tx_msg_ready: mpsc::UnboundedSender<(TConsensusSpec::Addr, HotstuffMessage)>,
-    rx_new_transactions: mpsc::Receiver<TransactionId>,
-    message_buffer: MessageBuffer<TConsensusSpec::Addr>,
+    pub rx_new_transactions: mpsc::Receiver<TransactionId>,
+    pub message_buffer: MessageBuffer<TConsensusSpec::Addr>,
     shutdown: ShutdownSignal,
 }
 
@@ -73,32 +73,6 @@ where TConsensusSpec: ConsensusSpec
         }
     }
 
-    pub async fn next(&mut self, current_height: NodeHeight) -> IncomingMessageResult<TConsensusSpec::Addr> {
-        loop {
-            tokio::select! {
-                biased;
-
-                _ = self.shutdown.wait() => { break Ok(None); }
-
-                msg_or_sync = self.message_buffer.next(current_height) => {
-                    break msg_or_sync;
-                },
-
-                Some((from, msg)) = self.rx_hotstuff_message.recv() => {
-                    if let Err(err) = self.handle_hotstuff_message(current_height, from, msg).await {
-                        error!(target: LOG_TARGET, "Error handling message: {}", err);
-                    }
-                },
-
-                 Some(tx_id) = self.rx_new_transactions.recv() => {
-                    if let Err(err) = self.check_if_parked_blocks_ready(current_height, &tx_id).await {
-                        error!(target: LOG_TARGET, "Error checking parked blocks: {}", err);
-                    }
-                },
-            }
-        }
-    }
-
     pub async fn discard(&mut self) {
         loop {
             tokio::select! {
@@ -123,7 +97,15 @@ where TConsensusSpec: ConsensusSpec
     ) -> Result<(), HotStuffError> {
         match msg {
             HotstuffMessage::Proposal(msg) => {
-                self.process_proposal(current_height, msg).await?;
+                self.process_local_proposal(current_height, msg).await?;
+            },
+            HotstuffMessage::ForeignProposal(ref proposal) => {
+                self.check_proposal(proposal.block.clone()).await?;
+                self.tx_msg_ready
+                    .send((from, msg))
+                    .map_err(|_| HotStuffError::InternalChannelClosed {
+                        context: "tx_msg_ready in InboundMessageWorker::handle_hotstuff_message",
+                    })?;
             },
             msg => self
                 .tx_msg_ready
@@ -135,7 +117,19 @@ where TConsensusSpec: ConsensusSpec
         Ok(())
     }
 
-    async fn process_proposal(
+    async fn check_proposal(&self, block: Block) -> Result<Option<Block>, HotStuffError> {
+        check_hash_and_height(&block)?;
+        let committee_for_block = self
+            .epoch_manager
+            .get_committee_by_validator_public_key(block.epoch(), block.proposed_by())
+            .await?;
+        check_proposed_by_leader(&self.leader_strategy, &committee_for_block, &block)?;
+        check_signature(&block)?;
+        check_quorum_certificate::<TConsensusSpec>(&block, &self.vote_signing_service, &self.epoch_manager).await?;
+        self.handle_missing_transactions(block).await
+    }
+
+    async fn process_local_proposal(
         &self,
         current_height: NodeHeight,
         proposal: ProposalMessage,
@@ -160,16 +154,7 @@ where TConsensusSpec: ConsensusSpec
             return Ok(());
         }
 
-        check_hash_and_height(&block)?;
-        let committee_for_block = self
-            .epoch_manager
-            .get_committee_by_validator_public_key(block.epoch(), block.proposed_by())
-            .await?;
-        check_proposed_by_leader(&self.leader_strategy, &committee_for_block, &block)?;
-        check_signature(&block)?;
-        check_quorum_certificate::<TConsensusSpec>(&block, &self.vote_signing_service, &self.epoch_manager).await?;
-
-        let Some(ready_block) = self.handle_missing_transactions(block).await? else {
+        let Some(ready_block) = self.check_proposal(block).await? else {
             // Block not ready
             return Ok(());
         };
@@ -184,7 +169,7 @@ where TConsensusSpec: ConsensusSpec
         Ok(())
     }
 
-    async fn check_if_parked_blocks_ready(
+    pub async fn check_if_parked_blocks_ready(
         &self,
         current_height: NodeHeight,
         transaction_id: &TransactionId,
@@ -301,7 +286,7 @@ where TConsensusSpec: ConsensusSpec
     }
 }
 
-struct MessageBuffer<TAddr> {
+pub struct MessageBuffer<TAddr> {
     buffer: BTreeMap<NodeHeight, VecDeque<(TAddr, HotstuffMessage)>>,
     rx_msg_ready: mpsc::UnboundedReceiver<(TAddr, HotstuffMessage)>,
 }

dan_layer/consensus/src/hotstuff/on_ready_to_vote_on_local_block.rs

@@ -950,6 +950,7 @@ where TConsensusSpec: ConsensusSpec
             // committees and within committees are not different in terms of size, speed, etc.
             let diff_from_leader = (my_index + local_committee.len() - leader_index as usize) % local_committee.len();
             // f+1 nodes (always including the leader) send the proposal to the foreign committee
+            // if diff_from_leader <= (local_committee.len() - 1) / 3 + 1 {
             if diff_from_leader <= local_committee.len() / 3 {
                 self.proposer.broadcast_proposal_foreignly(&block).await?;
             }

dan_layer/consensus/src/hotstuff/on_receive_foreign_proposal.rs

@@ -69,13 +69,26 @@ where TConsensusSpec: ConsensusSpec
             .epoch_manager
             .get_committee_shard(block.epoch(), vn.shard_key)
             .await?;
+        let foreign_proposal = ForeignProposal::new(committee_shard.bucket(), *block.id());
+        if self
+            .store
+            .with_read_tx(|tx| ForeignProposal::exists(tx, &foreign_proposal))?
+        {
+            warn!(
+                target: LOG_TARGET,
+                "🔥 FOREIGN PROPOSAL: Already received proposal for block {}",
+                block.id(),
+            );
+            return Ok(());
+        }
+
         let local_shard = self.epoch_manager.get_local_committee_shard(block.epoch()).await?;
         self.validate_proposed_block(&from, &block, committee_shard.bucket(), local_shard.bucket())?;
         // Is this ok? Can foreign node send invalid block that should still increment the counter?
         self.foreign_receive_counter.increment(&committee_shard.bucket());
         self.store.with_write_tx(|tx| {
             self.foreign_receive_counter.save(tx)?;
-            ForeignProposal::new(committee_shard.bucket(), *block.id()).upsert(tx)?;
+            foreign_proposal.upsert(tx)?;
             self.on_receive_foreign_block(tx, &block, &committee_shard)
         })?;
 

dan_layer/consensus/src/hotstuff/worker.rs

@@ -222,13 +222,25 @@ where TConsensusSpec: ConsensusSpec
             );
 
             tokio::select! {
-                msg_or_sync = self.inbound_message_worker.next(current_height) => {
+                msg_or_sync = self.inbound_message_worker.message_buffer.next(current_height) => {
                     if let Err(e) = self.on_new_hs_message(msg_or_sync).await {
                         self.on_failure("on_new_hs_message", &e).await;
                         return Err(e);
                     }
                 },
 
+                Some((from, msg)) = self.inbound_message_worker.rx_hotstuff_message.recv() => {
+                    if let Err(err) = self.inbound_message_worker.handle_hotstuff_message(current_height, from, msg).await {
+                        error!(target: LOG_TARGET, "Error handling message: {}", err);
+                    }
+                },
+
+                Some(tx_id) = self.inbound_message_worker.rx_new_transactions.recv() => {
+                    if let Err(err) = self.inbound_message_worker.check_if_parked_blocks_ready(current_height, &tx_id).await {
+                        error!(target: LOG_TARGET, "Error checking parked blocks: {}", err);
+                    }
+                },
+
                 Ok(event) = epoch_manager_events.recv() => {
                     self.handle_epoch_manager_event(event).await?;
                 },

dan_layer/state_store_sqlite/src/reader.rs

@@ -32,6 +32,8 @@ use tari_dan_storage::{
         Block,
         BlockId,
         Command,
+        Decision,
+        Evidence,
         ForeignProposal,
         ForeignReceiveCounters,
         ForeignSendCounters,
@@ -61,7 +63,7 @@ use tari_utilities::ByteArray;
 
 use crate::{
     error::SqliteStorageError,
-    serialization::{deserialize_hex_try_from, deserialize_json, serialize_hex},
+    serialization::{deserialize_hex_try_from, deserialize_json, parse_from_string, serialize_hex},
     sql_models,
     sqlite_transaction::SqliteTransaction,
 };
@@ -95,7 +97,7 @@ impl<'a, TAddr> SqliteStateStoreReadTransaction<'a, TAddr> {
 }
 
 impl<'a, TAddr: NodeAddressable + Serialize + DeserializeOwned> SqliteStateStoreReadTransaction<'a, TAddr> {
-    pub(super) fn get_transaction_atom_state_updates_between_blocks<'i, ITx>(
+    pub fn get_transaction_atom_state_updates_between_blocks<'i, ITx>(
         &mut self,
         from_block_id: &BlockId,
         to_block_id: &BlockId,
@@ -1144,6 +1146,17 @@ impl<TAddr: NodeAddressable + Serialize + DeserializeOwned> StateStoreReadTransa
         Ok(count > 0)
     }
 
+    fn transaction_pool_get_all(&mut self) -> Result<Vec<TransactionPoolRecord>, StorageError> {
+        use crate::schema::transaction_pool;
+        let txs = transaction_pool::table
+            .get_results::<sql_models::TransactionPoolRecord>(self.connection())
+            .map_err(|e| SqliteStorageError::DieselError {
+                operation: "transaction_pool_get_all",
+                source: e,
+            })?;
+        txs.into_iter().map(|tx| tx.try_convert(None)).collect()
+    }
+
     fn transaction_pool_get_many_ready(&mut self, max_txs: usize) -> Result<Vec<TransactionPoolRecord>, StorageError> {
         use crate::schema::transaction_pool;
 
@@ -1178,14 +1191,48 @@ impl<TAddr: NodeAddressable + Serialize + DeserializeOwned> StateStoreReadTransa
             updates.len()
         );
 
+        let mut used_substates = HashSet::<ShardId>::new();
+        let mut processed_substates = HashMap::<TransactionId, HashSet<ShardId>>::new();
+        for (tx_id, update) in &updates {
+            if let Some(Decision::Abort) = update
+                .local_decision
+                .as_ref()
+                .map(|decision| parse_from_string(decision.as_str()))
+                .transpose()?
+            {
+                // The aborted transaction don't lock any substates
+                continue;
+            }
+            let evidence = deserialize_json::<Evidence>(&update.evidence)?;
+            let evidence = evidence.shards_iter().copied().collect::<HashSet<_>>();
+            processed_substates.insert(deserialize_hex_try_from(tx_id)?, evidence);
+        }
+
         ready_txs
             .into_iter()
             .filter_map(|rec| {
                 let maybe_update = updates.remove(&rec.transaction_id);
                 match rec.try_convert(maybe_update) {
                     Ok(rec) => {
                         if rec.is_ready() {
-                            Some(Ok(rec))
+                            let tx_substates: HashSet<ShardId> = rec
+                                .transaction()
+                                .evidence
+                                .shards_iter()
+                                .copied()
+                                .collect::<HashSet<_>>();
+                            if tx_substates.is_disjoint(&used_substates) &&
+                                processed_substates.iter().all(|(tx_id, substates)| {
+                                    tx_id == rec.transaction_id() || tx_substates.is_disjoint(substates)
+                                })
+                            {
+                                used_substates.extend(tx_substates);
+                                Some(Ok(rec))
+                            } else {
+                                // TODO: If we don't switch to "no version" transaction, then we can abort these here.
+                                // That also requires changes to the on_ready_to_vote_on_local_block
+                                None
+                            }
                         } else {
                             None
                         }

dan_layer/storage/src/consensus_models/transaction_pool.rs

@@ -9,6 +9,7 @@ use std::{
 };
 
 use log::*;
+use serde::Serialize;
 use tari_dan_common_types::{
     committee::CommitteeShard,
     optional::{IsNotFoundError, Optional},
@@ -25,7 +26,7 @@ use crate::{
 
 const _LOG_TARGET: &str = "tari::dan::storage::transaction_pool";
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
 pub enum TransactionPoolStage {
     /// Transaction has just come in and has never been proposed
     New,
@@ -215,7 +216,7 @@ impl<TStateStore: StateStore> TransactionPool<TStateStore> {
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize)]
 pub struct TransactionPoolRecord {
     transaction: TransactionAtom,
     stage: TransactionPoolStage,

dan_layer/storage/src/state_store/mod.rs

@@ -166,6 +166,7 @@ pub trait StateStoreReadTransaction {
         transaction_id: &TransactionId,
     ) -> Result<TransactionPoolRecord, StorageError>;
     fn transaction_pool_exists(&mut self, transaction_id: &TransactionId) -> Result<bool, StorageError>;
+    fn transaction_pool_get_all(&mut self) -> Result<Vec<TransactionPoolRecord>, StorageError>;
     fn transaction_pool_get_many_ready(&mut self, max_txs: usize) -> Result<Vec<TransactionPoolRecord>, StorageError>;
     fn transaction_pool_count(
         &mut self,