Lr/eqc voting

crates/example-types/src/storage_types.rs

@@ -42,6 +42,7 @@ pub struct TestStorageState<TYPES: NodeType> {
     proposals: BTreeMap<TYPES::View, Proposal<TYPES, QuorumProposal<TYPES>>>,
     high_qc: Option<hotshot_types::simple_certificate::QuorumCertificate<TYPES>>,
     action: TYPES::View,
+    epoch: TYPES::Epoch,
 }
 
 impl<TYPES: NodeType> Default for TestStorageState<TYPES> {
@@ -52,6 +53,7 @@ impl<TYPES: NodeType> Default for TestStorageState<TYPES> {
             proposals: BTreeMap::new(),
             high_qc: None,
             action: TYPES::View::genesis(),
+            epoch: TYPES::Epoch::genesis(),
         }
     }
 }
@@ -101,6 +103,9 @@ impl<TYPES: NodeType> TestStorage<TYPES> {
     pub async fn last_actioned_view(&self) -> TYPES::View {
         self.inner.read().await.action
     }
+    pub async fn last_actioned_epoch(&self) -> TYPES::Epoch {
+        self.inner.read().await.epoch
+    }
 }
 
 #[async_trait]

crates/hotshot/src/lib.rs

@@ -128,6 +128,9 @@ pub struct SystemContext<TYPES: NodeType, I: NodeImplementation<TYPES>, V: Versi
     /// The view to enter when first starting consensus
     start_view: TYPES::View,
 
+    /// The epoch to enter when first starting consensus
+    start_epoch: TYPES::Epoch,
+
     /// Access to the output event stream.
     output_event_stream: (Sender<Event<TYPES>>, InactiveReceiver<Event<TYPES>>),
 
@@ -171,6 +174,7 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES>, V: Versions> Clone
             consensus: self.consensus.clone(),
             instance_state: Arc::clone(&self.instance_state),
             start_view: self.start_view,
+            start_epoch: self.start_epoch,
             output_event_stream: self.output_event_stream.clone(),
             external_event_stream: self.external_event_stream.clone(),
             anchored_leaf: self.anchored_leaf.clone(),
@@ -345,6 +349,7 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES>, V: Versions> SystemContext<T
             private_key,
             config,
             start_view: initializer.start_view,
+            start_epoch: initializer.start_epoch,
             network,
             memberships: Arc::new(memberships),
             metrics: Arc::clone(&consensus_metrics),
@@ -375,7 +380,10 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES>, V: Versions> SystemContext<T
         #[allow(clippy::panic)]
         self.internal_event_stream
             .0
-            .broadcast_direct(Arc::new(HotShotEvent::ViewChange(self.start_view)))
+            .broadcast_direct(Arc::new(HotShotEvent::ViewChange(
+                self.start_view,
+                self.start_epoch,
+            )))
             .await
             .unwrap_or_else(|_| {
                 panic!(
@@ -967,6 +975,8 @@ pub struct HotShotInitializer<TYPES: NodeType> {
 
     /// Starting view number that should be equivelant to the view the node shut down with last.
     start_view: TYPES::View,
+    /// Starting epoch number that should be equivelant to the epoch the node shut down with last.
+    start_epoch: TYPES::Epoch,
     /// The view we last performed an action in.  An action is Proposing or voting for
     /// Either the quorum or DA.
     actioned_view: TYPES::View,
@@ -1000,6 +1010,7 @@ impl<TYPES: NodeType> HotShotInitializer<TYPES> {
             validated_state: Some(Arc::new(validated_state)),
             state_delta: Some(Arc::new(state_delta)),
             start_view: TYPES::View::new(0),
+            start_epoch: TYPES::Epoch::new(0),
             actioned_view: TYPES::View::new(0),
             saved_proposals: BTreeMap::new(),
             high_qc,
@@ -1023,6 +1034,7 @@ impl<TYPES: NodeType> HotShotInitializer<TYPES> {
         instance_state: TYPES::InstanceState,
         validated_state: Option<Arc<TYPES::ValidatedState>>,
         start_view: TYPES::View,
+        start_epoch: TYPES::Epoch,
         actioned_view: TYPES::View,
         saved_proposals: BTreeMap<TYPES::View, Proposal<TYPES, QuorumProposal<TYPES>>>,
         high_qc: QuorumCertificate<TYPES>,
@@ -1036,6 +1048,7 @@ impl<TYPES: NodeType> HotShotInitializer<TYPES> {
             validated_state,
             state_delta: None,
             start_view,
+            start_epoch,
             actioned_view,
             saved_proposals,
             high_qc,

crates/hotshot/src/tasks/mod.rs

@@ -10,6 +10,11 @@
 pub mod task_state;
 use std::{collections::BTreeMap, fmt::Debug, num::NonZeroUsize, sync::Arc, time::Duration};
 
+use crate::{
+    tasks::task_state::CreateTaskState, types::SystemContextHandle, ConsensusApi,
+    ConsensusMetricsValue, ConsensusTaskRegistry, HotShotConfig, HotShotInitializer,
+    MarketplaceConfig, Memberships, NetworkTaskRegistry, SignatureKey, SystemContext, Versions,
+};
 use async_broadcast::{broadcast, RecvError};
 use async_lock::RwLock;
 use async_trait::async_trait;
@@ -32,7 +37,7 @@ use hotshot_task_impls::{
     view_sync::ViewSyncTaskState,
 };
 use hotshot_types::{
-    consensus::Consensus,
+    consensus::{Consensus, OuterConsensus},
     constants::EVENT_CHANNEL_SIZE,
     message::{Message, UpgradeLock},
     traits::{
@@ -43,12 +48,6 @@ use hotshot_types::{
 use tokio::{spawn, time::sleep};
 use vbs::version::StaticVersionType;
 
-use crate::{
-    tasks::task_state::CreateTaskState, types::SystemContextHandle, ConsensusApi,
-    ConsensusMetricsValue, ConsensusTaskRegistry, HotShotConfig, HotShotInitializer,
-    MarketplaceConfig, Memberships, NetworkTaskRegistry, SignatureKey, SystemContext, Versions,
-};
-
 /// event for global event stream
 #[derive(Clone, Debug)]
 pub enum GlobalEvent {
@@ -200,7 +199,7 @@ pub fn add_network_event_task<
         quorum_membership,
         da_membership,
         storage: Arc::clone(&handle.storage()),
-        consensus: Arc::clone(&handle.consensus()),
+        consensus: OuterConsensus::new(handle.consensus()),
         upgrade_lock: handle.hotshot.upgrade_lock.clone(),
         transmit_tasks: BTreeMap::new(),
     };

crates/hotshot/src/tasks/task_state.rs

@@ -241,6 +241,7 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES>, V: Versions> CreateTaskState
             id: handle.hotshot.id,
             storage: Arc::clone(&handle.storage),
             upgrade_lock: handle.hotshot.upgrade_lock.clone(),
+            epoch_height: handle.hotshot.config.epoch_height,
         }
     }
 }
@@ -293,6 +294,7 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES>, V: Versions> CreateTaskState
             spawned_tasks: BTreeMap::new(),
             id: handle.hotshot.id,
             upgrade_lock: handle.hotshot.upgrade_lock.clone(),
+            epoch_height: handle.hotshot.config.epoch_height,
         }
     }
 }
@@ -324,6 +326,7 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES>, V: Versions> CreateTaskState
             consensus: OuterConsensus::new(consensus),
             id: handle.hotshot.id,
             upgrade_lock: handle.hotshot.upgrade_lock.clone(),
+            epoch_height: handle.hotshot.config.epoch_height,
         }
     }
 }

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

@@ -38,14 +38,19 @@ pub(crate) async fn handle_quorum_vote_recv<
     sender: &Sender<Arc<HotShotEvent<TYPES>>>,
     task_state: &mut ConsensusTaskState<TYPES, I, V>,
 ) -> Result<()> {
-    // Are we the leader for this view?
+    let is_vote_leaf_extended = task_state
+        .consensus
+        .read()
+        .await
+        .is_leaf_extended(vote.data.leaf_commit);
+    let we_are_leader = task_state
+        .quorum_membership
+        .leader(vote.view_number() + 1, task_state.cur_epoch)?
+        == task_state.public_key;
     ensure!(
-        task_state
-            .quorum_membership
-            .leader(vote.view_number() + 1, task_state.cur_epoch)?
-            == task_state.public_key,
+        is_vote_leaf_extended || we_are_leader,
         info!(
-            "We are not the leader for view {:?}",
+            "We are not the leader for view {:?} and this is not the last vote for eQC",
             vote.view_number() + 1
         )
     );
@@ -60,6 +65,7 @@ pub(crate) async fn handle_quorum_vote_recv<
         &event,
         sender,
         &task_state.upgrade_lock,
+        !is_vote_leaf_extended,
     )
     .await?;
 
@@ -99,6 +105,7 @@ pub(crate) async fn handle_timeout_vote_recv<
         &event,
         sender,
         &task_state.upgrade_lock,
+        true,
     )
     .await?;
 
@@ -113,9 +120,15 @@ pub(crate) async fn handle_view_change<
     V: Versions,
 >(
     new_view_number: TYPES::View,
+    epoch_number: TYPES::Epoch,
     sender: &Sender<Arc<HotShotEvent<TYPES>>>,
     task_state: &mut ConsensusTaskState<TYPES, I, V>,
 ) -> Result<()> {
+    if epoch_number > task_state.cur_epoch {
+        task_state.cur_epoch = epoch_number;
+        tracing::info!("Progress: entered epoch {:>6}", *epoch_number);
+    }
+
     ensure!(
         new_view_number > task_state.cur_view,
         "New view is not larger than the current view"
@@ -129,7 +142,6 @@ pub(crate) async fn handle_view_change<
     }
     // Move this node to the next view
     task_state.cur_view = new_view_number;
-
     task_state
         .consensus
         .write()

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

@@ -92,10 +92,13 @@ pub struct ConsensusTaskState<TYPES: NodeType, I: NodeImplementation<TYPES>, V:
 
     /// Lock for a decided upgrade
     pub upgrade_lock: UpgradeLock<TYPES, V>,
+
+    /// Number of blocks in an epoch, zero means there are no epochs
+    pub epoch_height: u64,
 }
 impl<TYPES: NodeType, I: NodeImplementation<TYPES>, V: Versions> ConsensusTaskState<TYPES, I, V> {
     /// Handles a consensus event received on the event stream
-    #[instrument(skip_all, fields(id = self.id, cur_view = *self.cur_view), name = "Consensus replica task", level = "error", target = "ConsensusTaskState")]
+    #[instrument(skip_all, fields(id = self.id, cur_view = *self.cur_view, cur_epoch = *self.cur_epoch), name = "Consensus replica task", level = "error", target = "ConsensusTaskState")]
     pub async fn handle(
         &mut self,
         event: Arc<HotShotEvent<TYPES>>,
@@ -116,8 +119,10 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES>, V: Versions> ConsensusTaskSt
                     tracing::debug!("Failed to handle TimeoutVoteRecv event; error = {e}");
                 }
             }
-            HotShotEvent::ViewChange(new_view_number) => {
-                if let Err(e) = handle_view_change(*new_view_number, &sender, self).await {
+            HotShotEvent::ViewChange(new_view_number, epoch_number) => {
+                if let Err(e) =
+                    handle_view_change(*new_view_number, *epoch_number, &sender, self).await
+                {
                     tracing::trace!("Failed to handle ViewChange event; error = {e}");
                 }
             }

crates/task-impls/src/da.rs

@@ -84,7 +84,7 @@ pub struct DaTaskState<TYPES: NodeType, I: NodeImplementation<TYPES>, V: Version
 
 impl<TYPES: NodeType, I: NodeImplementation<TYPES>, V: Versions> DaTaskState<TYPES, I, V> {
     /// main task event handler
-    #[instrument(skip_all, fields(id = self.id, view = *self.cur_view), name = "DA Main Task", level = "error", target = "DaTaskState")]
+    #[instrument(skip_all, fields(id = self.id, view = *self.cur_view, epoch = *self.cur_epoch), name = "DA Main Task", level = "error", target = "DaTaskState")]
     pub async fn handle(
         &mut self,
         event: Arc<HotShotEvent<TYPES>>,
@@ -282,12 +282,16 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES>, V: Versions> DaTaskState<TYP
                     &event,
                     &event_stream,
                     &self.upgrade_lock,
+                    true,
                 )
                 .await?;
             }
-            HotShotEvent::ViewChange(view) => {
-                let view = *view;
+            HotShotEvent::ViewChange(view, epoch) => {
+                if *epoch > self.cur_epoch {
+                    self.cur_epoch = *epoch;
+                }
 
+                let view = *view;
                 ensure!(
                     *self.cur_view < *view,
                     info!("Received a view change to an older view.")

crates/task-impls/src/events.rs

@@ -92,6 +92,8 @@ pub enum HotShotEvent<TYPES: NodeType> {
     QuorumProposalSend(Proposal<TYPES, QuorumProposal<TYPES>>, TYPES::SignatureKey),
     /// Send a quorum vote to the next leader; emitted by a replica in the consensus task after seeing a valid quorum proposal
     QuorumVoteSend(QuorumVote<TYPES>),
+    /// Broadcast a quorum vote to form an eQC; emitted by a replica in the consensus task after seeing a valid quorum proposal
+    ExtendedQuorumVoteSend(QuorumVote<TYPES>),
     /// A quorum proposal with the given parent leaf is validated.
     /// The full validation checks include:
     /// 1. The proposal is not for an old view
@@ -122,7 +124,7 @@ pub enum HotShotEvent<TYPES: NodeType> {
     /// The DA leader has collected enough votes to form a DAC; emitted by the DA leader in the DA task; sent to the entire network via the networking task
     DacSend(DaCertificate<TYPES>, TYPES::SignatureKey),
     /// The current view has changed; emitted by the replica in the consensus task or replica in the view sync task; received by almost all other tasks
-    ViewChange(TYPES::View),
+    ViewChange(TYPES::View, TYPES::Epoch),
     /// Timeout for the view sync protocol; emitted by a replica in the view sync task
     ViewSyncTimeout(TYPES::View, u64, ViewSyncPhase),
 
@@ -253,7 +255,9 @@ impl<TYPES: NodeType> HotShotEvent<TYPES> {
             | HotShotEvent::QuorumProposalPreliminarilyValidated(proposal) => {
                 Some(proposal.data.view_number())
             }
-            HotShotEvent::QuorumVoteSend(vote) => Some(vote.view_number()),
+            HotShotEvent::QuorumVoteSend(vote) | HotShotEvent::ExtendedQuorumVoteSend(vote) => {
+                Some(vote.view_number())
+            }
             HotShotEvent::DaProposalRecv(proposal, _)
             | HotShotEvent::DaProposalValidated(proposal, _)
             | HotShotEvent::DaProposalSend(proposal, _) => Some(proposal.data.view_number()),
@@ -294,7 +298,7 @@ impl<TYPES: NodeType> HotShotEvent<TYPES> {
             }
             HotShotEvent::QuorumProposalRequestSend(req, _)
             | HotShotEvent::QuorumProposalRequestRecv(req, _) => Some(req.view_number),
-            HotShotEvent::ViewChange(view_number)
+            HotShotEvent::ViewChange(view_number, _)
             | HotShotEvent::ViewSyncTimeout(view_number, _, _)
             | HotShotEvent::ViewSyncTrigger(view_number)
             | HotShotEvent::Timeout(view_number) => Some(*view_number),
@@ -324,6 +328,13 @@ impl<TYPES: NodeType> Display for HotShotEvent<TYPES> {
             HotShotEvent::QuorumVoteRecv(v) => {
                 write!(f, "QuorumVoteRecv(view_number={:?})", v.view_number())
             }
+            HotShotEvent::ExtendedQuorumVoteSend(v) => {
+                write!(
+                    f,
+                    "ExtendedQuorumVoteSend(view_number={:?})",
+                    v.view_number()
+                )
+            }
             HotShotEvent::TimeoutVoteRecv(v) => {
                 write!(f, "TimeoutVoteRecv(view_number={:?})", v.view_number())
             }
@@ -379,8 +390,11 @@ impl<TYPES: NodeType> Display for HotShotEvent<TYPES> {
             HotShotEvent::DacSend(cert, _) => {
                 write!(f, "DacSend(view_number={:?})", cert.view_number())
             }
-            HotShotEvent::ViewChange(view_number) => {
-                write!(f, "ViewChange(view_number={view_number:?})")
+            HotShotEvent::ViewChange(view_number, epoch_number) => {
+                write!(
+                    f,
+                    "ViewChange(view_number={view_number:?}, epoch_number={epoch_number:?})"
+                )
             }
             HotShotEvent::ViewSyncTimeout(view_number, _, _) => {
                 write!(f, "ViewSyncTimeout(view_number={view_number:?})")

crates/task-impls/src/helpers.rs

@@ -13,6 +13,7 @@ use async_broadcast::{InactiveReceiver, Receiver, SendError, Sender};
 use async_lock::RwLock;
 use committable::{Commitment, Committable};
 use hotshot_task::dependency::{Dependency, EventDependency};
+use hotshot_types::utils::epoch_from_block_number;
 use hotshot_types::{
     consensus::OuterConsensus,
     data::{Leaf, QuorumProposal, ViewChangeEvidence},
@@ -481,9 +482,29 @@ pub async fn validate_proposal_safety_and_liveness<
     // Create a positive vote if either liveness or safety check
     // passes.
 
-    // Liveness check.
     {
         let consensus_reader = validation_info.consensus.read().await;
+        // Epoch safety check:
+        // The proposal is safe if
+        // 1. the proposed block and the justify QC block belong to the same epoch or
+        // 2. the justify QC is the eQC for the previous block
+        let proposal_epoch =
+            epoch_from_block_number(proposed_leaf.height(), validation_info.epoch_height);
+        let justify_qc_epoch =
+            epoch_from_block_number(parent_leaf.height(), validation_info.epoch_height);
+        ensure!(
+            proposal_epoch == justify_qc_epoch
+                || consensus_reader.check_eqc(&proposed_leaf, &parent_leaf),
+            {
+                error!(
+                    "Failed epoch safety check \n Proposed leaf is {:?} \n justify QC leaf is {:?}",
+                    proposed_leaf.clone(),
+                    parent_leaf.clone(),
+                )
+            }
+        );
+
+        // Liveness check.
         let liveness_check = justify_qc.view_number() > consensus_reader.locked_view();
 
         // Safety check.