wip

Cargo.toml

@@ -181,7 +181,7 @@ httpmock = "0.6.8"
 humantime = "2.1.0"
 humantime-serde = "1.1.1"
 include_dir = "0.7.2"
-indexmap = "2.2.6"
+indexmap = "2.5.0"
 indoc = "1.0.6"
 itertools = "0.11.0"
 lazy_static = "1.4.0"

dan_layer/common_types/src/substate_address.rs

@@ -117,6 +117,11 @@ impl SubstateAddress {
         &self.0[..ObjectKey::LENGTH]
     }
 
+    pub fn to_object_key(&self) -> ObjectKey {
+        ObjectKey::try_from(self.object_key_bytes())
+            .expect("SubstateAddress: object_key_bytes must return valid ObjectKey bytes")
+    }
+
     pub fn to_version(&self) -> u32 {
         let mut buf = [0u8; size_of::<u32>()];
         buf.copy_from_slice(&self.0[ObjectKey::LENGTH..]);

dan_layer/consensus/src/hotstuff/block_change_set.rs

@@ -82,6 +82,7 @@ impl ProposedBlockChangeSet {
     }
 
     pub fn no_vote(&mut self) -> &mut Self {
+        // TODO: store a NO-VOTE reason for debugging purposes so that we can review it easily
         // This means no vote
         self.quorum_decision = None;
         // The remaining info discarded (not strictly necessary)
@@ -198,7 +199,7 @@ impl ProposedBlockChangeSet {
     ) -> Result<Option<TransactionPoolRecord>, TransactionPoolError> {
         self.transaction_changes
             .get(transaction_id)
-            .and_then(|change| change.next_update.as_ref().map(|u| &u.transaction))
+            .and_then(|change| change.next_update.as_ref().map(|u| u.transaction()))
             .cloned()
             .map(Ok)
             .or_else(|| {
@@ -218,7 +219,8 @@ impl ProposedBlockChangeSet {
             .entry(*transaction.transaction_id())
             .or_default();
 
-        change_mut.next_update = Some(TransactionPoolStatusUpdate { transaction });
+        let ready_now = transaction.is_ready_for_next_stage();
+        change_mut.next_update = Some(TransactionPoolStatusUpdate::new(transaction, ready_now));
         Ok(self)
     }
 }

dan_layer/consensus/src/hotstuff/foreign_proposal_processor.rs

@@ -12,7 +12,6 @@ use tari_dan_storage::{
         ForeignProposal,
         LeafBlock,
         LockedBlock,
-        ShardGroupEvidence,
         TransactionAtom,
         TransactionPoolRecord,
         TransactionPoolStage,
@@ -121,30 +120,15 @@ pub fn process_foreign_block<TTx: StateStoreReadTransaction>(
                     proposed_block_change_set.add_transaction_execution(exec)?;
                 }
 
-                // We need to add the justify QC to the evidence because the all prepare block could not include it
-                // yet
-                let foreign_evidence = atom
-                    .evidence
-                    .get(&foreign_committee_info.shard_group())
-                    .ok_or_else(|| ProposalValidationError::ForeignInvalidPledge {
-                        block_id: *block.id(),
-                        transaction_id: atom.id,
-                        details: format!(
-                            "Foreign proposal did not contain evidence for {}",
-                            foreign_committee_info.shard_group()
-                        ),
-                    })?;
-
                 // Update the transaction record with any new information provided by this foreign block
                 tx_rec
                     .evidence_mut()
-                    .update(foreign_committee_info.shard_group(), foreign_evidence)
+                    .update(&atom.evidence)
                     .add_prepare_qc_evidence(foreign_committee_info, *justify_qc.id());
                 tx_rec.set_remote_decision(remote_decision);
 
                 validate_and_add_pledges(
                     &tx_rec,
-                    foreign_evidence,
                     block.id(),
                     atom,
                     &mut block_pledge,
@@ -279,30 +263,15 @@ pub fn process_foreign_block<TTx: StateStoreReadTransaction>(
                     proposed_block_change_set.add_transaction_execution(exec)?;
                 }
 
-                // We need to add the justify QC to the evidence because the all prepare block could not include it
-                // yet
-                let foreign_evidence = atom
-                    .evidence
-                    .get(&foreign_committee_info.shard_group())
-                    .ok_or_else(|| ProposalValidationError::ForeignInvalidPledge {
-                        block_id: *block.id(),
-                        transaction_id: atom.id,
-                        details: format!(
-                            "Foreign proposal did not contain evidence for {}",
-                            foreign_committee_info.shard_group()
-                        ),
-                    })?;
-
                 // Update the transaction record with any new information provided by this foreign block
                 tx_rec
                     .evidence_mut()
-                    .update(foreign_committee_info.shard_group(), foreign_evidence)
+                    .update(&atom.evidence)
                     .add_accept_qc_evidence(foreign_committee_info, *justify_qc.id());
                 tx_rec.set_remote_decision(remote_decision);
 
                 validate_and_add_pledges(
                     &tx_rec,
-                    foreign_evidence,
                     block.id(),
                     atom,
                     &mut block_pledge,
@@ -312,11 +281,11 @@ pub fn process_foreign_block<TTx: StateStoreReadTransaction>(
                 )?;
 
                 // Good debug info
-                // tx_rec.evidence().iter().for_each(|(addr, ev)| {
-                //     let includes_local = local_committee_info.includes_substate_address(addr);
+                // tx_rec.evidence().iter().for_each(|(sg, ev)| {
+                //     let is_local = local_committee_info.shard_group() == *sg;
                 //     log::error!(
                 //         target: LOG_TARGET,
-                //         "🐞 LOCALACCEPT EVIDENCE (l={}, f={}) {}: {}", includes_local, !includes_local, addr, ev
+                //         "🐞 LOCALACCEPT EVIDENCE (l={}, f={}) {}: {}", is_local, !is_local, sg, ev
                 //     );
                 // });
 
@@ -354,6 +323,17 @@ pub fn process_foreign_block<TTx: StateStoreReadTransaction>(
                         }
                         proposed_block_change_set.set_next_transaction_update(tx_rec)?;
                     }
+                } else if tx_rec.current_stage().is_local_prepared() && tx_rec.is_ready_for_next_stage() {
+                    info!(
+                        target: LOG_TARGET,
+                        "🧩 FOREIGN PROPOSAL: Transaction is ready for propose ALL_PREPARED({}, {}) Local Stage: {}",
+                        tx_rec.transaction_id(),
+                        tx_rec.current_decision(),
+                        tx_rec.current_stage()
+                    );
+
+                    tx_rec.set_next_stage(TransactionPoolStage::LocalPrepared)?;
+                    proposed_block_change_set.set_next_transaction_update(tx_rec)?;
                 } else if tx_rec.current_stage().is_local_accepted() && tx_rec.is_ready_for_next_stage() {
                     info!(
                         target: LOG_TARGET,
@@ -364,7 +344,6 @@ pub fn process_foreign_block<TTx: StateStoreReadTransaction>(
                     );
 
                     tx_rec.set_next_stage(TransactionPoolStage::LocalAccepted)?;
-
                     proposed_block_change_set.set_next_transaction_update(tx_rec)?;
                 } else {
                     info!(
@@ -376,7 +355,6 @@ pub fn process_foreign_block<TTx: StateStoreReadTransaction>(
                         tx_rec.evidence().all_addresses_accepted()
                     );
                     // Still need to update the evidence
-
                     proposed_block_change_set.set_next_transaction_update(tx_rec)?;
                 }
             },
@@ -423,14 +401,27 @@ pub fn process_foreign_block<TTx: StateStoreReadTransaction>(
 
 fn validate_and_add_pledges(
     transaction: &TransactionPoolRecord,
-    evidence: &ShardGroupEvidence,
     foreign_block_id: &BlockId,
     atom: &TransactionAtom,
     block_pledge: &mut BlockPledge,
     foreign_committee_info: &CommitteeInfo,
     proposed_block_change_set: &mut ProposedBlockChangeSet,
     is_prepare_phase: bool,
 ) -> Result<(), HotStuffError> {
+    // We need to add the justify QC to the evidence because the prepare block should not include it
+    // yet
+    let evidence = atom
+        .evidence
+        .get(&foreign_committee_info.shard_group())
+        .ok_or_else(|| ProposalValidationError::ForeignInvalidPledge {
+            block_id: *foreign_block_id,
+            transaction_id: atom.id,
+            details: format!(
+                "Foreign proposal did not contain evidence for {}",
+                foreign_committee_info.shard_group()
+            ),
+        })?;
+
     #[allow(clippy::mutable_key_type)]
     match atom.decision {
         Decision::Commit => {

dan_layer/consensus/src/hotstuff/on_inbound_message.rs

@@ -148,8 +148,9 @@ pub struct NeedsSync<TAddr: NodeAddressable> {
 fn msg_epoch_and_height(msg: &HotstuffMessage) -> Option<EpochAndHeight> {
     match msg {
         HotstuffMessage::Proposal(msg) => Some((msg.block.epoch(), msg.block.height())),
-        // Votes for block 2, occur at current height 3
+        // Votes for block 2 occur in view 3
         HotstuffMessage::Vote(msg) => Some((msg.epoch, msg.block_height.saturating_add(NodeHeight(1)))),
+        HotstuffMessage::NewView(msg) => Some((msg.high_qc.epoch(), msg.new_height)),
         _ => None,
     }
 }