hotshot_task_impls/consensus/
mod.rs

1// Copyright (c) 2021-2024 Espresso Systems (espressosys.com)
2// This file is part of the HotShot repository.
3
4// You should have received a copy of the MIT License
5// along with the HotShot repository. If not, see <https://mit-license.org/>.
6
7use std::{sync::Arc, time::Instant};
8
9use async_broadcast::{Receiver, Sender};
10use async_trait::async_trait;
11use handlers::handle_epoch_root_quorum_vote_recv;
12use hotshot_task::task::TaskState;
13use hotshot_types::{
14    consensus::OuterConsensus,
15    epoch_membership::EpochMembershipCoordinator,
16    event::Event,
17    message::UpgradeLock,
18    simple_certificate::{NextEpochQuorumCertificate2, QuorumCertificate2, TimeoutCertificate2},
19    simple_vote::{HasEpoch, NextEpochQuorumVote2, QuorumVote2, TimeoutVote2},
20    traits::{
21        node_implementation::{ConsensusTime, NodeImplementation, NodeType, Versions},
22        signature_key::SignatureKey,
23        storage::Storage,
24    },
25    utils::{epoch_from_block_number, is_last_block},
26    vote::HasViewNumber,
27};
28use hotshot_utils::anytrace::*;
29use tokio::task::JoinHandle;
30use tracing::instrument;
31
32use self::handlers::{
33    handle_quorum_vote_recv, handle_timeout, handle_timeout_vote_recv, handle_view_change,
34};
35use crate::{
36    events::HotShotEvent,
37    helpers::{broadcast_view_change, validate_qc_and_next_epoch_qc},
38    vote_collection::{EpochRootVoteCollectorsMap, VoteCollectorsMap},
39};
40
41/// Event handlers for use in the `handle` method.
42mod handlers;
43
44/// Task state for the Consensus task.
45pub struct ConsensusTaskState<TYPES: NodeType, I: NodeImplementation<TYPES>, V: Versions> {
46    /// Our public key
47    pub public_key: TYPES::SignatureKey,
48
49    /// Our Private Key
50    pub private_key: <TYPES::SignatureKey as SignatureKey>::PrivateKey,
51
52    /// Immutable instance state
53    pub instance_state: Arc<TYPES::InstanceState>,
54
55    /// The underlying network
56    pub network: Arc<I::Network>,
57
58    /// Membership for Quorum Certs/votes
59    pub membership_coordinator: EpochMembershipCoordinator<TYPES>,
60
61    /// A map of `QuorumVote` collector tasks.
62    pub vote_collectors: VoteCollectorsMap<TYPES, QuorumVote2<TYPES>, QuorumCertificate2<TYPES>, V>,
63
64    /// A map of `EpochRootQuorumVote` collector tasks.
65    pub epoch_root_vote_collectors: EpochRootVoteCollectorsMap<TYPES, V>,
66
67    /// A map of `QuorumVote` collector tasks. They collect votes from the nodes in the next epoch.
68    pub next_epoch_vote_collectors: VoteCollectorsMap<
69        TYPES,
70        NextEpochQuorumVote2<TYPES>,
71        NextEpochQuorumCertificate2<TYPES>,
72        V,
73    >,
74
75    /// A map of `TimeoutVote` collector tasks.
76    pub timeout_vote_collectors:
77        VoteCollectorsMap<TYPES, TimeoutVote2<TYPES>, TimeoutCertificate2<TYPES>, V>,
78
79    /// The view number that this node is currently executing in.
80    pub cur_view: TYPES::View,
81
82    /// Timestamp this view starts at.
83    pub cur_view_time: i64,
84
85    /// The epoch number that this node is currently executing in.
86    pub cur_epoch: Option<TYPES::Epoch>,
87
88    /// Output events to application
89    pub output_event_stream: async_broadcast::Sender<Event<TYPES>>,
90
91    /// Timeout task handle
92    pub timeout_task: JoinHandle<()>,
93
94    /// View timeout from config.
95    pub timeout: u64,
96
97    /// A reference to the metrics trait.
98    pub consensus: OuterConsensus<TYPES>,
99
100    /// A reference to the storage trait.
101    pub storage: I::Storage,
102
103    /// The node's id
104    pub id: u64,
105
106    /// Lock for a decided upgrade
107    pub upgrade_lock: UpgradeLock<TYPES, V>,
108
109    /// Number of blocks in an epoch, zero means there are no epochs
110    pub epoch_height: u64,
111
112    /// The time this view started
113    pub view_start_time: Instant,
114
115    /// First view in which epoch version takes effect
116    pub first_epoch: Option<(TYPES::View, TYPES::Epoch)>,
117}
118
119impl<TYPES: NodeType, I: NodeImplementation<TYPES>, V: Versions> ConsensusTaskState<TYPES, I, V> {
120    /// Handles a consensus event received on the event stream
121    #[instrument(skip_all, fields(id = self.id, cur_view = *self.cur_view, cur_epoch = self.cur_epoch.map(|x| *x)), name = "Consensus replica task", level = "error", target = "ConsensusTaskState")]
122    pub async fn handle(
123        &mut self,
124        event: Arc<HotShotEvent<TYPES>>,
125        sender: Sender<Arc<HotShotEvent<TYPES>>>,
126    ) -> Result<()> {
127        match event.as_ref() {
128            HotShotEvent::QuorumVoteRecv(ref vote) => {
129                if let Err(e) =
130                    handle_quorum_vote_recv(vote, Arc::clone(&event), &sender, self).await
131                {
132                    tracing::debug!("Failed to handle QuorumVoteRecv event; error = {e}");
133                }
134            },
135            HotShotEvent::EpochRootQuorumVoteRecv(ref vote) => {
136                if let Err(e) =
137                    handle_epoch_root_quorum_vote_recv(vote, Arc::clone(&event), &sender, self)
138                        .await
139                {
140                    tracing::debug!("Failed to handle EpochRootQuorumVoteRecv event; error = {e}");
141                }
142            },
143            HotShotEvent::TimeoutVoteRecv(ref vote) => {
144                if let Err(e) =
145                    handle_timeout_vote_recv(vote, Arc::clone(&event), &sender, self).await
146                {
147                    tracing::debug!("Failed to handle TimeoutVoteRecv event; error = {e}");
148                }
149            },
150            HotShotEvent::SetFirstEpoch(view, epoch) => {
151                self.first_epoch = Some((*view, *epoch));
152            },
153            HotShotEvent::ViewChange(new_view_number, epoch_number) => {
154                if let Err(e) =
155                    handle_view_change(*new_view_number, *epoch_number, &sender, self).await
156                {
157                    tracing::trace!("Failed to handle ViewChange event; error = {e}");
158                }
159                self.view_start_time = Instant::now();
160            },
161            HotShotEvent::Timeout(view_number, epoch) => {
162                if let Err(e) = handle_timeout(*view_number, *epoch, &sender, self).await {
163                    tracing::debug!("Failed to handle Timeout event; error = {e}");
164                }
165            },
166            HotShotEvent::ExtendedQc2Formed(eqc) => {
167                let cert_view = eqc.view_number();
168                let Some(cert_block_number) = eqc.data.block_number else {
169                    tracing::error!("Received extended QC but no block number");
170                    return Ok(());
171                };
172                let cert_epoch = epoch_from_block_number(cert_block_number, self.epoch_height);
173                tracing::error!("Formed Extended QC for view {cert_view} and epoch {cert_epoch}.");
174                // Transition to the new epoch by sending ViewChange
175                let next_epoch = TYPES::Epoch::new(cert_epoch + 1);
176                broadcast_view_change(&sender, cert_view + 1, Some(next_epoch), self.first_epoch)
177                    .await;
178                tracing::info!("Entering new epoch: {next_epoch}");
179                tracing::info!(
180                    "Stake table for epoch {}:\n\n{:?}",
181                    next_epoch,
182                    self.membership_coordinator
183                        .stake_table_for_epoch(Some(next_epoch))
184                        .await?
185                        .stake_table()
186                        .await
187                );
188                tracing::info!(
189                    "Stake table for epoch {}:\n\n{:?}",
190                    next_epoch + 1,
191                    self.membership_coordinator
192                        .stake_table_for_epoch(Some(next_epoch + 1))
193                        .await?
194                        .stake_table()
195                        .await
196                );
197            },
198            HotShotEvent::ExtendedQcRecv(high_qc, next_epoch_high_qc, _) => {
199                if !high_qc
200                    .data
201                    .block_number
202                    .is_some_and(|bn| is_last_block(bn, self.epoch_height))
203                {
204                    tracing::warn!("Received extended QC but we can't verify the leaf is extended");
205                    return Ok(());
206                }
207                if let Err(e) = validate_qc_and_next_epoch_qc(
208                    high_qc,
209                    Some(next_epoch_high_qc),
210                    &self.consensus,
211                    &self.membership_coordinator,
212                    &self.upgrade_lock,
213                    self.epoch_height,
214                )
215                .await
216                {
217                    tracing::error!("Received invalid extended QC: {e}");
218                    return Ok(());
219                }
220
221                let mut consensus_writer = self.consensus.write().await;
222                let high_qc_updated = consensus_writer.update_high_qc(high_qc.clone()).is_ok();
223                let next_high_qc_updated = consensus_writer
224                    .update_next_epoch_high_qc(next_epoch_high_qc.clone())
225                    .is_ok();
226                drop(consensus_writer);
227
228                self.storage
229                    .update_high_qc2(high_qc.clone())
230                    .await
231                    .map_err(|_| warn!("Failed to update high QC"))?;
232                self.storage
233                    .update_next_epoch_high_qc2(next_epoch_high_qc.clone())
234                    .await
235                    .map_err(|_| warn!("Failed to update next epoch high QC"))?;
236
237                tracing::debug!(
238                    "Received Extended QC for view {} and epoch {:?}.",
239                    high_qc.view_number(),
240                    high_qc.epoch()
241                );
242                if high_qc_updated || next_high_qc_updated {
243                    // Send ViewChange indicating new view and new epoch.
244                    let next_epoch = high_qc.data.epoch().map(|x| x + 1);
245                    tracing::info!("Entering new epoch: {next_epoch:?}");
246                    broadcast_view_change(
247                        &sender,
248                        high_qc.view_number() + 1,
249                        next_epoch,
250                        self.first_epoch,
251                    )
252                    .await;
253                }
254            },
255            _ => {},
256        }
257
258        Ok(())
259    }
260}
261
262#[async_trait]
263impl<TYPES: NodeType, I: NodeImplementation<TYPES>, V: Versions> TaskState
264    for ConsensusTaskState<TYPES, I, V>
265{
266    type Event = HotShotEvent<TYPES>;
267
268    async fn handle_event(
269        &mut self,
270        event: Arc<Self::Event>,
271        sender: &Sender<Arc<Self::Event>>,
272        _receiver: &Receiver<Arc<Self::Event>>,
273    ) -> Result<()> {
274        self.handle(event, sender.clone()).await
275    }
276
277    /// Joins all subtasks.
278    fn cancel_subtasks(&mut self) {
279        // Cancel the old timeout task
280        std::mem::replace(&mut self.timeout_task, tokio::spawn(async {})).abort();
281    }
282}
hotshot_task_impls/consensus/mod.rs

hotshot_task_impls/consensus/
mod.rs
