Hotstuff 2 (#3877)

* removing some unused stuff and adding HighQC message + event

* Update high qc events, add helper to send

* Send high Qc on view change

* wait for highest qc in the proposal task

* fn stub

* working on fn

* new decide rule and locked qc rule for hs2

* Merge main

* add failures test

* actually use new decide rule

* add test verifying the new decide rule is one shorter

* reduce epoch chain len by 1

* fix waiting and validate QC

* use proper timeout

* more accurate high qc waiting logic

* gate by versios before waiting for QC

* Fix GC for epochs

* fix version gating

crates/hotshot/src/tasks/task_state.rs

@@ -270,6 +270,7 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES>, V: Versions> CreateTaskState
             formed_upgrade_certificate: None,
             upgrade_lock: handle.hotshot.upgrade_lock.clone(),
             epoch_height: handle.hotshot.config.epoch_height,
+            highest_qc: handle.hotshot.consensus.read().await.high_qc().clone(),
         }
     }
 }

crates/task-impls/src/consensus/handlers.rs

@@ -26,6 +26,7 @@ use crate::{
     consensus::Versions, events::HotShotEvent, helpers::broadcast_event,
     vote_collection::handle_vote,
 };
+use vbs::version::StaticVersionType;
 
 /// Handle a `QuorumVoteRecv` event.
 pub(crate) async fn handle_quorum_vote_recv<
@@ -112,6 +113,30 @@ pub(crate) async fn handle_timeout_vote_recv<
     Ok(())
 }
 
+/// Send an event to the next leader containing the highest QC we have
+/// This is a necessary part of HotStuff 2 but not the original HotStuff
+///
+/// #Errors
+/// Returns and error if we can't get the version or the version doesn't
+/// yet support HS 2
+pub async fn send_high_qc<TYPES: NodeType, V: Versions, I: NodeImplementation<TYPES>>(
+    new_view_number: TYPES::View,
+    sender: &Sender<Arc<HotShotEvent<TYPES>>>,
+    task_state: &mut ConsensusTaskState<TYPES, I, V>,
+) -> Result<()> {
+    let version = task_state.upgrade_lock.version(new_view_number).await?;
+    ensure!(
+        version >= V::Epochs::VERSION,
+        debug!("HotStuff 2 updgrade not yet in effect")
+    );
+    let high_qc = task_state.consensus.read().await.high_qc().clone();
+    let leader = task_state
+        .quorum_membership
+        .leader(new_view_number, TYPES::Epoch::new(0))?;
+    broadcast_event(Arc::new(HotShotEvent::HighQcSend(high_qc, leader)), sender).await;
+    Ok(())
+}
+
 /// Handle a `ViewChange` event.
 #[instrument(skip_all)]
 pub(crate) async fn handle_view_change<
@@ -140,6 +165,15 @@ pub(crate) async fn handle_view_change<
     if *old_view_number / 100 != *new_view_number / 100 {
         tracing::info!("Progress: entered view {:>6}", *new_view_number);
     }
+
+    // Send our high qc to the next leader immediately upon finishing a view.
+    // Part of HotStuff 2
+    let _ = send_high_qc(new_view_number, sender, task_state)
+        .await
+        .inspect_err(|e| {
+            tracing::debug!("High QC sending failed with error: {:?}", e);
+        });
+
     // Move this node to the next view
     task_state.cur_view = new_view_number;
     task_state

crates/task-impls/src/events.rs

@@ -236,6 +236,12 @@ pub enum HotShotEvent<TYPES: NodeType> {
         TYPES::SignatureKey,
         Proposal<TYPES, VidDisperseShare<TYPES>>,
     ),
+
+    /// A replica send us a High QC
+    HighQcRecv(QuorumCertificate<TYPES>, TYPES::SignatureKey),
+
+    /// Send our HighQC to the next leader, should go to the same leader as our vote
+    HighQcSend(QuorumCertificate<TYPES>, TYPES::SignatureKey),
 }
 
 impl<TYPES: NodeType> HotShotEvent<TYPES> {
@@ -311,6 +317,9 @@ impl<TYPES: NodeType> HotShotEvent<TYPES> {
             | HotShotEvent::VidRequestRecv(request, _) => Some(request.view),
             HotShotEvent::VidResponseSend(_, _, proposal)
             | HotShotEvent::VidResponseRecv(_, proposal) => Some(proposal.data.view_number),
+            HotShotEvent::HighQcRecv(qc, _) | HotShotEvent::HighQcSend(qc, _) => {
+                Some(qc.view_number())
+            }
         }
     }
 }
@@ -569,6 +578,12 @@ impl<TYPES: NodeType> Display for HotShotEvent<TYPES> {
                     proposal.data.view_number
                 )
             }
+            HotShotEvent::HighQcRecv(qc, _) => {
+                write!(f, "HighQcRecv(view_number={:?}", qc.view_number())
+            }
+            HotShotEvent::HighQcSend(qc, _) => {
+                write!(f, "HighQcSend(view_number={:?}", qc.view_number())
+            }
         }
     }
 }

crates/task-impls/src/helpers.rs

@@ -9,7 +9,7 @@ use std::{
     sync::Arc,
 };
 
-use async_broadcast::{InactiveReceiver, Receiver, SendError, Sender};
+use async_broadcast::{Receiver, SendError, Sender};
 use async_lock::RwLock;
 use committable::{Commitment, Committable};
 use hotshot_task::dependency::{Dependency, EventDependency};
@@ -192,6 +192,89 @@ impl<TYPES: NodeType + Default> Default for LeafChainTraversalOutcome<TYPES> {
     }
 }
 
+/// calculate the new decided leaf chain based on the rules of hostuff 2
+///
+/// # Panics
+/// Can't actually panic
+pub async fn decide_from_proposal_2<TYPES: NodeType>(
+    proposal: &QuorumProposal<TYPES>,
+    consensus: OuterConsensus<TYPES>,
+    existing_upgrade_cert: Arc<RwLock<Option<UpgradeCertificate<TYPES>>>>,
+    public_key: &TYPES::SignatureKey,
+) -> LeafChainTraversalOutcome<TYPES> {
+    let mut res = LeafChainTraversalOutcome::default();
+    let consensus_reader = consensus.read().await;
+    let proposed_leaf = Leaf::from_quorum_proposal(proposal);
+    res.new_locked_view_number = Some(proposed_leaf.justify_qc().view_number());
+
+    // If we don't have the proposals parent return early
+    let Some(parent_info) = consensus_reader.parent_leaf_info(&proposed_leaf, public_key) else {
+        return res;
+    };
+    // Get the parents parent and check if it's consecutive in view to the parent, if so we can decided
+    // the grandparents view.  If not we're done.
+    let Some(grand_parent_info) = consensus_reader.parent_leaf_info(&parent_info.leaf, public_key)
+    else {
+        return res;
+    };
+    if grand_parent_info.leaf.view_number() + 1 != parent_info.leaf.view_number() {
+        return res;
+    }
+    res.new_decide_qc = Some(parent_info.leaf.justify_qc().clone());
+    let decided_view_number = grand_parent_info.leaf.view_number();
+    res.new_decided_view_number = Some(decided_view_number);
+    // We've reached decide, now get the leaf chain all the way back to the last decided view, not including it.
+    let old_anchor_view = consensus_reader.last_decided_view();
+    let mut current_leaf_info = Some(grand_parent_info);
+    let existing_upgrade_cert_reader = existing_upgrade_cert.read().await;
+    let mut txns = HashSet::new();
+    while current_leaf_info
+        .as_ref()
+        .is_some_and(|info| info.leaf.view_number() > old_anchor_view)
+    {
+        // unwrap is safe, we just checked that he option is some
+        let info = &mut current_leaf_info.unwrap();
+        // Check if there's a new upgrade certificate available.
+        if let Some(cert) = info.leaf.upgrade_certificate() {
+            if info.leaf.upgrade_certificate() != *existing_upgrade_cert_reader {
+                if cert.data.decide_by < decided_view_number {
+                    tracing::warn!("Failed to decide an upgrade certificate in time. Ignoring.");
+                } else {
+                    tracing::info!("Reached decide on upgrade certificate: {:?}", cert);
+                    res.decided_upgrade_cert = Some(cert.clone());
+                }
+            }
+        }
+
+        res.leaf_views.push(info.clone());
+        // If the block payload is available for this leaf, include it in
+        // the leaf chain that we send to the client.
+        if let Some(encoded_txns) = consensus_reader
+            .saved_payloads()
+            .get(&info.leaf.view_number())
+        {
+            let payload =
+                BlockPayload::from_bytes(encoded_txns, info.leaf.block_header().metadata());
+
+            info.leaf.fill_block_payload_unchecked(payload);
+        }
+
+        if let Some(ref payload) = info.leaf.block_payload() {
+            for txn in payload.transaction_commitments(info.leaf.block_header().metadata()) {
+                txns.insert(txn);
+            }
+        }
+
+        current_leaf_info = consensus_reader.parent_leaf_info(&info.leaf, public_key);
+    }
+
+    if !txns.is_empty() {
+        res.included_txns = Some(txns);
+    }
+
+    res
+}
+
 /// Ascends the leaf chain by traversing through the parent commitments of the proposal. We begin
 /// by obtaining the parent view, and if we are in a chain (i.e. the next view from the parent is
 /// one view newer), then we begin attempting to form the chain. This is a direct impl from
@@ -344,7 +427,7 @@ pub async fn decide_from_proposal<TYPES: NodeType>(
 pub(crate) async fn parent_leaf_and_state<TYPES: NodeType, V: Versions>(
     next_proposal_view_number: TYPES::View,
     event_sender: &Sender<Arc<HotShotEvent<TYPES>>>,
-    event_receiver: &InactiveReceiver<Arc<HotShotEvent<TYPES>>>,
+    event_receiver: &Receiver<Arc<HotShotEvent<TYPES>>>,
     quorum_membership: Arc<TYPES::Membership>,
     public_key: TYPES::SignatureKey,
     private_key: <TYPES::SignatureKey as SignatureKey>::PrivateKey,
@@ -370,7 +453,7 @@ pub(crate) async fn parent_leaf_and_state<TYPES: NodeType, V: Versions>(
         let _ = fetch_proposal(
             parent_view_number,
             event_sender.clone(),
-            event_receiver.activate_cloned(),
+            event_receiver.clone(),
             quorum_membership,
             consensus.clone(),
             public_key.clone(),

crates/task-impls/src/network.rs

@@ -77,7 +77,7 @@ impl<TYPES: NodeType> NetworkMessageTaskState<TYPES> {
                         GeneralConsensusMessage::ProposalRequested(req, sig) => {
                             HotShotEvent::QuorumProposalRequestRecv(req, sig)
                         }
-                        GeneralConsensusMessage::LeaderProposalAvailable(proposal) => {
+                        GeneralConsensusMessage::ProposalResponse(proposal) => {
                             HotShotEvent::QuorumProposalResponseRecv(proposal)
                         }
                         GeneralConsensusMessage::Vote(vote) => {
@@ -114,6 +114,7 @@ impl<TYPES: NodeType> NetworkMessageTaskState<TYPES> {
                             tracing::error!("Received upgrade vote!");
                             HotShotEvent::UpgradeVoteRecv(message)
                         }
+                        GeneralConsensusMessage::HighQC(qc) => HotShotEvent::HighQcRecv(qc, sender),
                     },
                     SequencingMessage::Da(da_message) => match da_message {
                         DaConsensusMessage::DaProposal(proposal) => {
@@ -428,7 +429,7 @@ impl<
             HotShotEvent::QuorumProposalResponseSend(sender_key, proposal) => Some((
                 sender_key.clone(),
                 MessageKind::<TYPES>::from_consensus_message(SequencingMessage::General(
-                    GeneralConsensusMessage::LeaderProposalAvailable(proposal),
+                    GeneralConsensusMessage::ProposalResponse(proposal),
                 )),
                 TransmitType::Direct(sender_key),
             )),