hotshot_types/
consensus.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
7//! Provides the core consensus types
8
9use std::{
10    collections::{BTreeMap, HashMap, HashSet},
11    mem::ManuallyDrop,
12    ops::{Deref, DerefMut},
13    sync::Arc,
14};
15
16use alloy_primitives::U256;
17use async_lock::{RwLock, RwLockReadGuard, RwLockUpgradableReadGuard, RwLockWriteGuard};
18use committable::{Commitment, Committable};
19use hotshot_utils::anytrace::*;
20use tracing::instrument;
21use vec1::Vec1;
22
23pub use crate::utils::{View, ViewInner};
24use crate::{
25    constants::EPOCH_PARTICIPATION_HISTORY,
26    data::{
27        EpochNumber, Leaf2, QuorumProposalWrapper, VidCommitment, VidDisperse,
28        VidDisperseAndDuration, VidDisperseShare, ViewNumber,
29    },
30    epoch_membership::EpochMembershipCoordinator,
31    error::HotShotError,
32    event::{HotShotAction, LeafInfo},
33    message::{Proposal, UpgradeLock},
34    simple_certificate::{
35        DaCertificate2, LightClientStateUpdateCertificateV2, NextEpochQuorumCertificate2,
36        QuorumCertificate2,
37    },
38    simple_vote::HasEpoch,
39    stake_table::{HSStakeTable, StakeTableEntries},
40    traits::{
41        BlockPayload, ValidatedState,
42        block_contents::{BlockHeader, BuilderFee},
43        metrics::{Counter, Gauge, Histogram, Metrics, NoMetrics},
44        node_implementation::NodeType,
45        signature_key::{SignatureKey, StakeTableEntryType},
46    },
47    utils::{
48        BuilderCommitment, LeafCommitment, StateAndDelta, Terminator, epoch_from_block_number,
49        is_epoch_root, is_epoch_transition, is_last_block, is_transition_block,
50        option_epoch_from_block_number,
51    },
52    vote::{Certificate, HasViewNumber},
53};
54
55/// A type alias for `HashMap<Commitment<T>, T>`
56pub type CommitmentMap<T> = HashMap<Commitment<T>, T>;
57
58/// A type alias for `BTreeMap<T::Time, HashMap<T::SignatureKey, BTreeMap<T::Epoch, Proposal<T, VidDisperseShare<T>>>>>`
59pub type VidShares<TYPES> = BTreeMap<
60    ViewNumber,
61    HashMap<
62        <TYPES as NodeType>::SignatureKey,
63        BTreeMap<Option<EpochNumber>, Proposal<TYPES, VidDisperseShare<TYPES>>>,
64    >,
65>;
66
67/// Type alias for consensus state wrapped in a lock.
68pub type LockedConsensusState<TYPES> = Arc<RwLock<Consensus<TYPES>>>;
69
70/// A thin wrapper around `LockedConsensusState` that helps debugging locks
71#[derive(Clone, Debug)]
72pub struct OuterConsensus<TYPES: NodeType> {
73    /// Inner `LockedConsensusState`
74    pub inner_consensus: LockedConsensusState<TYPES>,
75}
76
77impl<TYPES: NodeType> OuterConsensus<TYPES> {
78    /// Create a new instance of `OuterConsensus`, hopefully uniquely named
79    pub fn new(consensus: LockedConsensusState<TYPES>) -> Self {
80        Self {
81            inner_consensus: consensus,
82        }
83    }
84
85    /// Locks inner consensus for reading and leaves debug traces
86    #[instrument(skip_all, target = "OuterConsensus")]
87    pub async fn read(&self) -> ConsensusReadLockGuard<'_, TYPES> {
88        tracing::trace!("Trying to acquire read lock on consensus");
89        let ret = self.inner_consensus.read().await;
90        tracing::trace!("Acquired read lock on consensus");
91        ConsensusReadLockGuard::new(ret)
92    }
93
94    /// Locks inner consensus for writing and leaves debug traces
95    #[instrument(skip_all, target = "OuterConsensus")]
96    pub async fn write(&self) -> ConsensusWriteLockGuard<'_, TYPES> {
97        tracing::trace!("Trying to acquire write lock on consensus");
98        let ret = self.inner_consensus.write().await;
99        tracing::trace!("Acquired write lock on consensus");
100        ConsensusWriteLockGuard::new(ret)
101    }
102
103    /// Tries to acquire write lock on inner consensus and leaves debug traces
104    #[instrument(skip_all, target = "OuterConsensus")]
105    pub fn try_write(&self) -> Option<ConsensusWriteLockGuard<'_, TYPES>> {
106        tracing::trace!("Trying to acquire write lock on consensus");
107        let ret = self.inner_consensus.try_write();
108        if let Some(guard) = ret {
109            tracing::trace!("Acquired write lock on consensus");
110            Some(ConsensusWriteLockGuard::new(guard))
111        } else {
112            tracing::trace!("Failed to acquire write lock");
113            None
114        }
115    }
116
117    /// Acquires upgradable read lock on inner consensus and leaves debug traces
118    #[instrument(skip_all, target = "OuterConsensus")]
119    pub async fn upgradable_read(&self) -> ConsensusUpgradableReadLockGuard<'_, TYPES> {
120        tracing::trace!("Trying to acquire upgradable read lock on consensus");
121        let ret = self.inner_consensus.upgradable_read().await;
122        tracing::trace!("Acquired upgradable read lock on consensus");
123        ConsensusUpgradableReadLockGuard::new(ret)
124    }
125
126    /// Tries to acquire read lock on inner consensus and leaves debug traces
127    #[instrument(skip_all, target = "OuterConsensus")]
128    pub fn try_read(&self) -> Option<ConsensusReadLockGuard<'_, TYPES>> {
129        tracing::trace!("Trying to acquire read lock on consensus");
130        let ret = self.inner_consensus.try_read();
131        if let Some(guard) = ret {
132            tracing::trace!("Acquired read lock on consensus");
133            Some(ConsensusReadLockGuard::new(guard))
134        } else {
135            tracing::trace!("Failed to acquire read lock");
136            None
137        }
138    }
139}
140
141/// A thin wrapper around `RwLockReadGuard` for `Consensus` that leaves debug traces when the lock is freed
142pub struct ConsensusReadLockGuard<'a, TYPES: NodeType> {
143    /// Inner `RwLockReadGuard`
144    lock_guard: RwLockReadGuard<'a, Consensus<TYPES>>,
145}
146
147impl<'a, TYPES: NodeType> ConsensusReadLockGuard<'a, TYPES> {
148    /// Creates a new instance of `ConsensusReadLockGuard` with the same name as parent `OuterConsensus`
149    #[must_use]
150    pub fn new(lock_guard: RwLockReadGuard<'a, Consensus<TYPES>>) -> Self {
151        Self { lock_guard }
152    }
153}
154
155impl<TYPES: NodeType> Deref for ConsensusReadLockGuard<'_, TYPES> {
156    type Target = Consensus<TYPES>;
157    fn deref(&self) -> &Self::Target {
158        &self.lock_guard
159    }
160}
161
162impl<TYPES: NodeType> Drop for ConsensusReadLockGuard<'_, TYPES> {
163    #[instrument(skip_all, target = "ConsensusReadLockGuard")]
164    fn drop(&mut self) {
165        tracing::trace!("Read lock on consensus dropped");
166    }
167}
168
169/// A thin wrapper around `RwLockWriteGuard` for `Consensus` that leaves debug traces when the lock is freed
170pub struct ConsensusWriteLockGuard<'a, TYPES: NodeType> {
171    /// Inner `RwLockWriteGuard`
172    lock_guard: RwLockWriteGuard<'a, Consensus<TYPES>>,
173}
174
175impl<'a, TYPES: NodeType> ConsensusWriteLockGuard<'a, TYPES> {
176    /// Creates a new instance of `ConsensusWriteLockGuard` with the same name as parent `OuterConsensus`
177    #[must_use]
178    pub fn new(lock_guard: RwLockWriteGuard<'a, Consensus<TYPES>>) -> Self {
179        Self { lock_guard }
180    }
181}
182
183impl<TYPES: NodeType> Deref for ConsensusWriteLockGuard<'_, TYPES> {
184    type Target = Consensus<TYPES>;
185    fn deref(&self) -> &Self::Target {
186        &self.lock_guard
187    }
188}
189
190impl<TYPES: NodeType> DerefMut for ConsensusWriteLockGuard<'_, TYPES> {
191    fn deref_mut(&mut self) -> &mut Self::Target {
192        &mut self.lock_guard
193    }
194}
195
196impl<TYPES: NodeType> Drop for ConsensusWriteLockGuard<'_, TYPES> {
197    #[instrument(skip_all, target = "ConsensusWriteLockGuard")]
198    fn drop(&mut self) {
199        tracing::debug!("Write lock on consensus dropped");
200    }
201}
202
203/// A thin wrapper around `RwLockUpgradableReadGuard` for `Consensus` that leaves debug traces when the lock is freed or upgraded
204pub struct ConsensusUpgradableReadLockGuard<'a, TYPES: NodeType> {
205    /// Inner `RwLockUpgradableReadGuard`
206    lock_guard: ManuallyDrop<RwLockUpgradableReadGuard<'a, Consensus<TYPES>>>,
207    /// A helper bool to indicate whether inner lock has been unsafely taken or not
208    taken: bool,
209}
210
211impl<'a, TYPES: NodeType> ConsensusUpgradableReadLockGuard<'a, TYPES> {
212    /// Creates a new instance of `ConsensusUpgradableReadLockGuard` with the same name as parent `OuterConsensus`
213    #[must_use]
214    pub fn new(lock_guard: RwLockUpgradableReadGuard<'a, Consensus<TYPES>>) -> Self {
215        Self {
216            lock_guard: ManuallyDrop::new(lock_guard),
217            taken: false,
218        }
219    }
220
221    /// Upgrades the inner `RwLockUpgradableReadGuard` and leaves debug traces
222    #[instrument(skip_all, target = "ConsensusUpgradableReadLockGuard")]
223    #[allow(unused_assignments)] // `taken` is read in Drop impl
224    pub async fn upgrade(mut guard: Self) -> ConsensusWriteLockGuard<'a, TYPES> {
225        let inner_guard = unsafe { ManuallyDrop::take(&mut guard.lock_guard) };
226        guard.taken = true;
227        tracing::debug!("Trying to upgrade upgradable read lock on consensus");
228        let ret = RwLockUpgradableReadGuard::upgrade(inner_guard).await;
229        tracing::debug!("Upgraded upgradable read lock on consensus");
230        ConsensusWriteLockGuard::new(ret)
231    }
232}
233
234impl<TYPES: NodeType> Deref for ConsensusUpgradableReadLockGuard<'_, TYPES> {
235    type Target = Consensus<TYPES>;
236
237    fn deref(&self) -> &Self::Target {
238        &self.lock_guard
239    }
240}
241
242impl<TYPES: NodeType> Drop for ConsensusUpgradableReadLockGuard<'_, TYPES> {
243    #[instrument(skip_all, target = "ConsensusUpgradableReadLockGuard")]
244    fn drop(&mut self) {
245        if !self.taken {
246            unsafe { ManuallyDrop::drop(&mut self.lock_guard) }
247            tracing::debug!("Upgradable read lock on consensus dropped");
248        }
249    }
250}
251
252/// A bundle of views that we have most recently performed some action
253#[derive(Debug, Clone, Copy)]
254struct HotShotActionViews {
255    /// View we last proposed in to the Quorum
256    proposed: ViewNumber,
257    /// View we last voted in for a QuorumProposal
258    voted: ViewNumber,
259    /// View we last proposed to the DA committee
260    da_proposed: ViewNumber,
261    /// View we lasted voted for DA proposal
262    da_vote: ViewNumber,
263}
264
265impl Default for HotShotActionViews {
266    fn default() -> Self {
267        let genesis = ViewNumber::genesis();
268        Self {
269            proposed: genesis,
270            voted: genesis,
271            da_proposed: genesis,
272            da_vote: genesis,
273        }
274    }
275}
276impl HotShotActionViews {
277    /// Create HotShotActionViews from a view number
278    fn from_view(view: ViewNumber) -> Self {
279        Self {
280            proposed: view,
281            voted: view,
282            da_proposed: view,
283            da_vote: view,
284        }
285    }
286}
287
288type ValidatorParticipationMap<TYPES> = HashMap<<TYPES as NodeType>::SignatureKey, (u64, u64)>;
289
290/// Tracks proposal and vote participation per validator across epochs.
291///
292/// Owned by whichever consensus implementation is active: legacy
293/// [`Consensus`] feeds the same underlying trackers through its own
294/// methods, while the new-protocol coordinator owns one of these
295/// directly. Both serve the node participation API from it.
296pub struct ParticipationTracker<TYPES: NodeType> {
297    validator: ValidatorParticipation<TYPES>,
298    vote: VoteParticipation<TYPES>,
299}
300
301impl<TYPES: NodeType> Default for ParticipationTracker<TYPES> {
302    fn default() -> Self {
303        let (stake_table, success_threshold) = VoteParticipation::unresolved_stake_table();
304        Self {
305            validator: ValidatorParticipation::new(),
306            vote: VoteParticipation::new(stake_table, success_threshold, None),
307        }
308    }
309}
310
311impl<TYPES: NodeType> ParticipationTracker<TYPES> {
312    pub fn new(membership: &EpochMembershipCoordinator<TYPES>, epoch: EpochNumber) -> Self {
313        let (stake_table, success_threshold) =
314            resolve_participation_stake_table(membership, Some(epoch));
315        Self {
316            validator: ValidatorParticipation::new_in_epoch(epoch),
317            vote: VoteParticipation::new(stake_table, success_threshold, Some(epoch)),
318        }
319    }
320
321    pub fn leader_proposed(&mut self, leader: TYPES::SignatureKey, epoch: EpochNumber) {
322        self.validator.update_participation(leader, epoch, true);
323    }
324
325    pub fn leader_missed(&mut self, leader: TYPES::SignatureKey, epoch: EpochNumber) {
326        self.validator.update_participation(leader, epoch, false);
327    }
328
329    pub fn on_view_changed(&mut self, epoch: EpochNumber) {
330        self.validator.update_participation_epoch(epoch);
331    }
332
333    pub fn on_leaf_decided(
334        &mut self,
335        leaf: &Leaf2<TYPES>,
336        membership: &EpochMembershipCoordinator<TYPES>,
337    ) {
338        if let Err(err) = track_decided_qc_participation(
339            &leaf.justify_qc(),
340            membership,
341            &mut self.validator,
342            &mut self.vote,
343        ) {
344            tracing::warn!(%err, "failed to update vote participation epoch");
345        }
346    }
347
348    pub fn current_proposal_participation(&self) -> HashMap<TYPES::SignatureKey, f64> {
349        self.validator.current_proposal_participation()
350    }
351
352    pub fn proposal_participation(&self, epoch: EpochNumber) -> HashMap<TYPES::SignatureKey, f64> {
353        self.validator.proposal_participation(epoch)
354    }
355
356    pub fn current_vote_participation(
357        &self,
358    ) -> HashMap<<TYPES::SignatureKey as SignatureKey>::VerificationKeyType, f64> {
359        self.vote.current_vote_participation()
360    }
361
362    pub fn vote_participation(
363        &self,
364        epoch: EpochNumber,
365    ) -> HashMap<<TYPES::SignatureKey as SignatureKey>::VerificationKeyType, f64> {
366        self.vote.vote_participation(Some(epoch))
367    }
368}
369
370#[derive(Debug, Clone)]
371struct ValidatorParticipation<TYPES: NodeType> {
372    epoch: EpochNumber,
373    /// Current epoch participation by key maps key -> (num leader, num times proposed)
374    current_epoch_participation: ValidatorParticipationMap<TYPES>,
375
376    /// Previous epochs participation ordered by epoch number, maps epoch -> key -> (num leader, num times proposed)
377    previous_epoch_participation: BTreeMap<EpochNumber, ValidatorParticipationMap<TYPES>>,
378}
379
380impl<TYPES: NodeType> ValidatorParticipation<TYPES> {
381    fn new() -> Self {
382        Self::new_in_epoch(EpochNumber::genesis())
383    }
384
385    fn new_in_epoch(epoch: EpochNumber) -> Self {
386        Self {
387            epoch,
388            current_epoch_participation: HashMap::new(),
389            previous_epoch_participation: BTreeMap::new(),
390        }
391    }
392
393    fn update_participation(
394        &mut self,
395        key: TYPES::SignatureKey,
396        epoch: EpochNumber,
397        proposed: bool,
398    ) {
399        let participation = match epoch.cmp(&self.epoch) {
400            std::cmp::Ordering::Greater => {
401                self.update_participation_epoch(epoch);
402                &mut self.current_epoch_participation
403            },
404            std::cmp::Ordering::Equal => &mut self.current_epoch_participation,
405            std::cmp::Ordering::Less => {
406                let Some(archived) = self.previous_epoch_participation.get_mut(&epoch) else {
407                    return;
408                };
409                archived
410            },
411        };
412        let entry = participation.entry(key).or_insert((0, 0));
413        if proposed {
414            entry.1 += 1;
415        }
416        entry.0 += 1;
417    }
418
419    fn update_participation_epoch(&mut self, epoch: EpochNumber) {
420        if epoch <= self.epoch {
421            return;
422        }
423        self.previous_epoch_participation
424            .insert(self.epoch, self.current_epoch_participation.clone());
425
426        self.previous_epoch_participation =
427            self.previous_epoch_participation
428                .split_off(&EpochNumber::new(
429                    self.epoch.saturating_sub(EPOCH_PARTICIPATION_HISTORY),
430                ));
431
432        self.epoch = epoch;
433        self.current_epoch_participation = HashMap::new();
434    }
435
436    fn current_proposal_participation(&self) -> HashMap<TYPES::SignatureKey, f64> {
437        self.current_epoch_participation
438            .iter()
439            .map(|(key, (leader, proposed))| {
440                (
441                    key.clone(),
442                    if *leader == 0 {
443                        0.0
444                    } else {
445                        *proposed as f64 / *leader as f64
446                    },
447                )
448            })
449            .collect()
450    }
451    fn proposal_participation(&self, epoch: EpochNumber) -> HashMap<TYPES::SignatureKey, f64> {
452        let tracked_participation = if epoch == self.epoch {
453            self.current_epoch_participation.clone()
454        } else {
455            self.previous_epoch_participation
456                .get(&epoch)
457                .unwrap_or(&HashMap::new())
458                .clone()
459        };
460
461        tracked_participation
462            .iter()
463            .map(|(key, (leader, proposed))| {
464                (
465                    key.clone(),
466                    if *leader == 0 {
467                        0.0
468                    } else {
469                        *proposed as f64 / *leader as f64
470                    },
471                )
472            })
473            .collect()
474    }
475
476    fn current_epoch(&self) -> EpochNumber {
477        self.epoch
478    }
479}
480
481type VoteParticipationMap<TYPES> = (
482    HashMap<<<TYPES as NodeType>::SignatureKey as SignatureKey>::VerificationKeyType, u64>,
483    u64,
484);
485
486#[derive(Clone, Debug)]
487struct VoteParticipation<TYPES: NodeType> {
488    /// Current epoch
489    epoch: Option<EpochNumber>,
490
491    /// Current stake_table
492    stake_table: HSStakeTable<TYPES>,
493
494    /// Success threshold
495    success_threshold: U256,
496
497    /// Set of views in the current epoch
498    view_set: HashSet<ViewNumber>,
499
500    /// Number of views in the current epoch
501    current_epoch_num_views: u64,
502
503    /// Current epoch participation by key maps key -> num times voted
504    current_epoch_participation:
505        HashMap<<TYPES::SignatureKey as SignatureKey>::VerificationKeyType, u64>,
506
507    /// Previous epochs participation ordered by epoch number, maps epoch -> key -> num times voted
508    previous_epoch_participation: BTreeMap<Option<EpochNumber>, VoteParticipationMap<TYPES>>,
509}
510
511impl<TYPES: NodeType> VoteParticipation<TYPES> {
512    fn new(
513        stake_table: HSStakeTable<TYPES>,
514        success_threshold: U256,
515        epoch: Option<EpochNumber>,
516    ) -> Self {
517        let current_epoch_participation: HashMap<_, _> = stake_table
518            .iter()
519            .map({
520                |peer_config| {
521                    (
522                        peer_config
523                            .stake_table_entry
524                            .public_key()
525                            .to_verification_key(),
526                        0u64,
527                    )
528                }
529            })
530            .collect();
531        Self {
532            epoch,
533            stake_table,
534            success_threshold,
535            view_set: HashSet::new(),
536            current_epoch_num_views: 0u64,
537            current_epoch_participation,
538            previous_epoch_participation: BTreeMap::new(),
539        }
540    }
541
542    fn unresolved_stake_table() -> (HSStakeTable<TYPES>, U256) {
543        (HSStakeTable::default(), U256::MAX)
544    }
545
546    fn stake_table_unresolved(&self) -> bool {
547        self.stake_table.is_empty()
548    }
549
550    fn reseed_stake_table(&mut self, stake_table: HSStakeTable<TYPES>, threshold: U256) {
551        self.current_epoch_participation = stake_table
552            .iter()
553            .map(|peer_config| {
554                (
555                    peer_config
556                        .stake_table_entry
557                        .public_key()
558                        .to_verification_key(),
559                    0u64,
560                )
561            })
562            .collect();
563        self.stake_table = stake_table;
564        self.success_threshold = threshold;
565    }
566
567    fn update_participation(&mut self, qc: QuorumCertificate2<TYPES>) -> Result<()> {
568        ensure!(
569            qc.epoch() == self.epoch,
570            info!(
571                "Incorrect epoch while updating vote participation, current epoch: {:?}, QC epoch \
572                 {:?}",
573                self.epoch,
574                qc.epoch()
575            )
576        );
577        ensure!(
578            !self.view_set.contains(&qc.view_number()),
579            info!(
580                "Participation for view {} already updated",
581                qc.view_number()
582            )
583        );
584        let signers = qc
585            .signers(
586                &StakeTableEntries::<TYPES>::from(self.stake_table.clone()).0,
587                self.success_threshold,
588            )
589            .context(|e| warn!("Tracing signers: {e}"))?;
590        for vk in signers {
591            let Some(votes) = self.current_epoch_participation.get_mut(&vk) else {
592                bail!(warn!(
593                    "Trying to update vote participation for unknown key: {:?}",
594                    vk
595                ));
596            };
597            *votes += 1;
598        }
599        self.view_set.insert(qc.view_number());
600        self.current_epoch_num_views += 1;
601        Ok(())
602    }
603
604    fn update_participation_epoch(
605        &mut self,
606        stake_table: HSStakeTable<TYPES>,
607        success_threshold: U256,
608        epoch: Option<EpochNumber>,
609    ) -> Result<()> {
610        ensure!(
611            epoch >= self.epoch,
612            warn!(
613                "New epoch less than current epoch while updating vote participation epoch, \
614                 current epoch: {:?}, new epoch {:?}",
615                self.epoch, epoch
616            )
617        );
618        // Same epoch, do nothing
619        if epoch == self.epoch {
620            return Ok(());
621        }
622
623        self.previous_epoch_participation.insert(
624            self.epoch,
625            (
626                self.current_epoch_participation.clone(),
627                self.current_epoch_num_views,
628            ),
629        );
630
631        self.previous_epoch_participation = self.previous_epoch_participation.split_off(
632            &self
633                .epoch
634                .map(|e| EpochNumber::new(e.saturating_sub(EPOCH_PARTICIPATION_HISTORY))),
635        );
636
637        self.previous_epoch_participation.insert(
638            self.epoch,
639            (
640                self.current_epoch_participation.clone(),
641                self.current_epoch_num_views,
642            ),
643        );
644
645        self.previous_epoch_participation = self.previous_epoch_participation.split_off(
646            &self
647                .epoch
648                .map(|e| EpochNumber::new(e.saturating_sub(EPOCH_PARTICIPATION_HISTORY))),
649        );
650
651        self.epoch = epoch;
652        self.current_epoch_num_views = 0;
653        self.view_set = HashSet::new();
654        let current_epoch_participation: HashMap<_, _> = stake_table
655            .iter()
656            .map({
657                |peer_config| {
658                    (
659                        peer_config
660                            .stake_table_entry
661                            .public_key()
662                            .to_verification_key(),
663                        0u64,
664                    )
665                }
666            })
667            .collect();
668        self.current_epoch_participation = current_epoch_participation;
669        self.stake_table = stake_table;
670        self.success_threshold = success_threshold;
671        Ok(())
672    }
673
674    fn current_vote_participation(
675        &self,
676    ) -> HashMap<<TYPES::SignatureKey as SignatureKey>::VerificationKeyType, f64> {
677        self.current_epoch_participation
678            .iter()
679            .map(|(key, votes)| {
680                (
681                    key.clone(),
682                    Self::calculate_ratio(votes, self.current_epoch_num_views),
683                )
684            })
685            .collect()
686    }
687    fn vote_participation(
688        &self,
689        epoch: Option<EpochNumber>,
690    ) -> HashMap<<TYPES::SignatureKey as SignatureKey>::VerificationKeyType, f64> {
691        let (participation, num_views) = if epoch == self.epoch {
692            (
693                self.current_epoch_participation.clone(),
694                self.current_epoch_num_views,
695            )
696        } else {
697            self.previous_epoch_participation
698                .get(&epoch)
699                .unwrap_or(&(HashMap::new(), 0))
700                .clone()
701        };
702
703        if num_views == 0 {
704            return HashMap::new();
705        }
706
707        participation
708            .iter()
709            .map(|(key, votes)| (key.clone(), Self::calculate_ratio(votes, num_views)))
710            .collect()
711    }
712
713    fn calculate_ratio(num_votes: &u64, total_views: u64) -> f64 {
714        if total_views == 0 {
715            0.0
716        } else {
717            *num_votes as f64 / total_views as f64
718        }
719    }
720
721    fn current_epoch(&self) -> Option<EpochNumber> {
722        self.epoch
723    }
724}
725
726fn resolve_participation_stake_table<TYPES: NodeType>(
727    membership: &EpochMembershipCoordinator<TYPES>,
728    epoch: Option<EpochNumber>,
729) -> (HSStakeTable<TYPES>, U256) {
730    match membership.stake_table_for_epoch(epoch) {
731        Ok(m) => (
732            HSStakeTable::from_iter(m.stake_table()),
733            m.success_threshold(),
734        ),
735        Err(err) => {
736            tracing::warn!(?epoch, %err, "no stake table for participation tracking");
737            VoteParticipation::unresolved_stake_table()
738        },
739    }
740}
741
742fn track_decided_qc_participation<TYPES: NodeType>(
743    qc: &QuorumCertificate2<TYPES>,
744    membership: &EpochMembershipCoordinator<TYPES>,
745    validator: &mut ValidatorParticipation<TYPES>,
746    vote: &mut VoteParticipation<TYPES>,
747) -> Result<()> {
748    let qc_epoch = qc.epoch();
749    if let Some(epoch) = qc_epoch
750        && epoch > validator.current_epoch()
751    {
752        validator.update_participation_epoch(epoch);
753    }
754    if qc_epoch > vote.current_epoch() {
755        let (stake_table, success_threshold) =
756            resolve_participation_stake_table(membership, qc_epoch);
757        vote.update_participation_epoch(stake_table, success_threshold, qc_epoch)
758            .context(warn!("Updating vote participation"))?;
759    } else if qc_epoch == vote.current_epoch()
760        && vote.stake_table_unresolved()
761        && let Ok(m) = membership.stake_table_for_epoch(qc_epoch)
762    {
763        vote.reseed_stake_table(
764            HSStakeTable::from_iter(m.stake_table()),
765            m.success_threshold(),
766        );
767    }
768    if let Err(e) = vote.update_participation(qc.clone()) {
769        tracing::warn!("Failed to update vote participation: {e}");
770    }
771    Ok(())
772}
773
774/// A reference to the consensus algorithm
775///
776/// This will contain the state of all rounds.
777#[derive(derive_more::Debug, Clone)]
778pub struct Consensus<TYPES: NodeType> {
779    /// The validated states that are currently loaded in memory.
780    validated_state_map: BTreeMap<ViewNumber, View<TYPES>>,
781
782    /// All the VID shares we've received for current and future views.
783    vid_shares: VidShares<TYPES>,
784
785    /// All the DA certs we've received for current and future views.
786    /// view -> DA cert
787    saved_da_certs: HashMap<ViewNumber, DaCertificate2<TYPES>>,
788
789    /// View number that is currently on.
790    cur_view: ViewNumber,
791
792    /// Epoch number that is currently on.
793    cur_epoch: Option<EpochNumber>,
794
795    /// Last proposals we sent out, None if we haven't proposed yet.
796    /// Prevents duplicate proposals, and can be served to those trying to catchup
797    last_proposals: BTreeMap<ViewNumber, Proposal<TYPES, QuorumProposalWrapper<TYPES>>>,
798
799    /// last view had a successful decide event
800    last_decided_view: ViewNumber,
801
802    /// The `locked_qc` view number
803    locked_view: ViewNumber,
804
805    /// Map of leaf hash -> leaf
806    /// - contains undecided leaves
807    /// - includes the MOST RECENT decided leaf
808    saved_leaves: CommitmentMap<Leaf2<TYPES>>,
809
810    /// Bundle of views which we performed the most recent action
811    /// visibible to the network.  Actions are votes and proposals
812    /// for DA and Quorum
813    last_actions: HotShotActionViews,
814
815    /// Saved payloads.
816    ///
817    /// Encoded transactions for every view if we got a payload for that view.
818    saved_payloads: BTreeMap<ViewNumber, Arc<PayloadWithMetadata<TYPES>>>,
819
820    /// the highqc per spec
821    high_qc: QuorumCertificate2<TYPES>,
822
823    /// The high QC for the next epoch
824    next_epoch_high_qc: Option<NextEpochQuorumCertificate2<TYPES>>,
825
826    /// Track the participation of each validator in the current epoch and previous epoch
827    validator_participation: ValidatorParticipation<TYPES>,
828
829    /// Track the vote participation of each node in the current epoch and previous epoch
830    vote_participation: VoteParticipation<TYPES>,
831
832    /// A reference to the metrics trait
833    pub metrics: Arc<ConsensusMetricsValue>,
834
835    /// Number of blocks in an epoch, zero means there are no epochs
836    pub epoch_height: u64,
837
838    /// Number of iterations for the DRB calculation, taken from HotShotConfig
839    pub drb_difficulty: u64,
840
841    /// Number of iterations for the DRB calculation post-difficulty upgrade, taken from HotShotConfig
842    pub drb_upgrade_difficulty: u64,
843
844    /// The transition QC for the current epoch
845    transition_qc: Option<(
846        QuorumCertificate2<TYPES>,
847        NextEpochQuorumCertificate2<TYPES>,
848    )>,
849
850    /// The highest block number that we have seen
851    pub highest_block: u64,
852    /// The light client state update certificate
853    pub state_cert: Option<LightClientStateUpdateCertificateV2<TYPES>>,
854}
855
856/// This struct holds a payload and its metadata
857#[derive(Debug, Clone, Hash, Eq, PartialEq)]
858pub struct PayloadWithMetadata<TYPES: NodeType> {
859    pub payload: TYPES::BlockPayload,
860    pub metadata: <TYPES::BlockPayload as BlockPayload<TYPES>>::Metadata,
861}
862
863/// Contains several `ConsensusMetrics` that we're interested in from the consensus interfaces
864#[derive(Clone, Debug)]
865pub struct ConsensusMetricsValue {
866    /// The number of last synced block height
867    pub last_synced_block_height: Box<dyn Gauge>,
868    /// The number of last decided view
869    pub last_decided_view: Box<dyn Gauge>,
870    /// The number of the last voted view
871    pub last_voted_view: Box<dyn Gauge>,
872    /// Number of timestamp for the last decided time
873    pub last_decided_time: Box<dyn Gauge>,
874    /// The current view
875    pub current_view: Box<dyn Gauge>,
876    /// Number of views that are in-flight since the last decided view
877    pub number_of_views_since_last_decide: Box<dyn Gauge>,
878    /// Number of views that are in-flight since the last anchor view
879    pub number_of_views_per_decide_event: Box<dyn Histogram>,
880    /// Duration of views as leader
881    pub view_duration_as_leader: Box<dyn Histogram>,
882    /// Number of invalid QCs we've seen since the last commit.
883    pub invalid_qc: Box<dyn Gauge>,
884    /// Number of outstanding transactions
885    pub outstanding_transactions: Box<dyn Gauge>,
886    /// Memory size in bytes of the serialized transactions still outstanding
887    pub outstanding_transactions_memory_size: Box<dyn Gauge>,
888    /// Number of views that timed out
889    pub number_of_timeouts: Box<dyn Counter>,
890    /// Number of views that timed out as leader
891    pub number_of_timeouts_as_leader: Box<dyn Counter>,
892    /// The number of empty blocks that have been proposed
893    pub number_of_empty_blocks_proposed: Box<dyn Counter>,
894    /// Number of events in the hotshot event queue
895    pub internal_event_queue_len: Box<dyn Gauge>,
896    /// Time from proposal creation to decide time
897    pub proposal_to_decide_time: Box<dyn Histogram>,
898    /// Time from proposal received to proposal creation
899    pub previous_proposal_to_proposal_time: Box<dyn Histogram>,
900    /// Finalized bytes per view
901    pub finalized_bytes: Box<dyn Histogram>,
902    /// The duration of the validate and apply header
903    pub validate_and_apply_header_duration: Box<dyn Histogram>,
904    /// The duration of update leaf
905    pub update_leaf_duration: Box<dyn Histogram>,
906    /// The time it took to calculate the disperse
907    pub vid_disperse_duration: Box<dyn Histogram>,
908}
909
910impl ConsensusMetricsValue {
911    /// Create a new instance of this [`ConsensusMetricsValue`] struct, setting all the counters and gauges
912    #[must_use]
913    pub fn new(metrics: &dyn Metrics) -> Self {
914        Self {
915            last_synced_block_height: metrics
916                .create_gauge(String::from("last_synced_block_height"), None),
917            last_decided_view: metrics.create_gauge(String::from("last_decided_view"), None),
918            last_voted_view: metrics.create_gauge(String::from("last_voted_view"), None),
919            last_decided_time: metrics.create_gauge(String::from("last_decided_time"), None),
920            current_view: metrics.create_gauge(String::from("current_view"), None),
921            number_of_views_since_last_decide: metrics
922                .create_gauge(String::from("number_of_views_since_last_decide"), None),
923            number_of_views_per_decide_event: metrics
924                .create_histogram(String::from("number_of_views_per_decide_event"), None),
925            view_duration_as_leader: metrics
926                .create_histogram(String::from("view_duration_as_leader"), None),
927            invalid_qc: metrics.create_gauge(String::from("invalid_qc"), None),
928            outstanding_transactions: metrics
929                .create_gauge(String::from("outstanding_transactions"), None),
930            outstanding_transactions_memory_size: metrics
931                .create_gauge(String::from("outstanding_transactions_memory_size"), None),
932            number_of_timeouts: metrics.create_counter(String::from("number_of_timeouts"), None),
933            number_of_timeouts_as_leader: metrics
934                .create_counter(String::from("number_of_timeouts_as_leader"), None),
935            number_of_empty_blocks_proposed: metrics
936                .create_counter(String::from("number_of_empty_blocks_proposed"), None),
937            internal_event_queue_len: metrics
938                .create_gauge(String::from("internal_event_queue_len"), None),
939            proposal_to_decide_time: metrics
940                .create_histogram(String::from("proposal_to_decide_time"), None),
941            previous_proposal_to_proposal_time: metrics
942                .create_histogram(String::from("previous_proposal_to_proposal_time"), None),
943            finalized_bytes: metrics.create_histogram(String::from("finalized_bytes"), None),
944            validate_and_apply_header_duration: metrics.create_histogram(
945                String::from("validate_and_apply_header_duration"),
946                Some("seconds".to_string()),
947            ),
948            update_leaf_duration: metrics.create_histogram(
949                String::from("update_leaf_duration"),
950                Some("seconds".to_string()),
951            ),
952            vid_disperse_duration: metrics.create_histogram(
953                String::from("vid_disperse_duration"),
954                Some("seconds".to_string()),
955            ),
956        }
957    }
958}
959
960impl Default for ConsensusMetricsValue {
961    fn default() -> Self {
962        Self::new(&*NoMetrics::boxed())
963    }
964}
965
966impl<TYPES: NodeType> Consensus<TYPES> {
967    /// Constructor.
968    #[allow(clippy::too_many_arguments)]
969    pub fn new(
970        validated_state_map: BTreeMap<ViewNumber, View<TYPES>>,
971        vid_shares: Option<VidShares<TYPES>>,
972        cur_view: ViewNumber,
973        cur_epoch: Option<EpochNumber>,
974        locked_view: ViewNumber,
975        last_decided_view: ViewNumber,
976        last_actioned_view: ViewNumber,
977        last_proposals: BTreeMap<ViewNumber, Proposal<TYPES, QuorumProposalWrapper<TYPES>>>,
978        saved_leaves: CommitmentMap<Leaf2<TYPES>>,
979        saved_payloads: BTreeMap<ViewNumber, Arc<PayloadWithMetadata<TYPES>>>,
980        high_qc: QuorumCertificate2<TYPES>,
981        next_epoch_high_qc: Option<NextEpochQuorumCertificate2<TYPES>>,
982        metrics: Arc<ConsensusMetricsValue>,
983        epoch_height: u64,
984        state_cert: Option<LightClientStateUpdateCertificateV2<TYPES>>,
985        drb_difficulty: u64,
986        drb_upgrade_difficulty: u64,
987        stake_table: HSStakeTable<TYPES>,
988        success_threshold: U256,
989    ) -> Self {
990        let transition_qc = if let Some(ref next_epoch_high_qc) = next_epoch_high_qc {
991            if high_qc
992                .data
993                .block_number
994                .is_some_and(|bn| is_transition_block(bn, epoch_height))
995            {
996                if high_qc.data.leaf_commit == next_epoch_high_qc.data.leaf_commit {
997                    Some((high_qc.clone(), next_epoch_high_qc.clone()))
998                } else {
999                    tracing::error!("Next epoch high QC has different leaf commit to high QC");
1000                    None
1001                }
1002            } else {
1003                None
1004            }
1005        } else {
1006            None
1007        };
1008        Consensus {
1009            validated_state_map,
1010            vid_shares: vid_shares.unwrap_or_default(),
1011            saved_da_certs: HashMap::new(),
1012            cur_view,
1013            cur_epoch,
1014            last_decided_view,
1015            last_proposals,
1016            last_actions: HotShotActionViews::from_view(last_actioned_view),
1017            locked_view,
1018            saved_leaves,
1019            saved_payloads,
1020            high_qc,
1021            next_epoch_high_qc,
1022            metrics,
1023            epoch_height,
1024            transition_qc,
1025            highest_block: 0,
1026            state_cert,
1027            drb_difficulty,
1028            validator_participation: ValidatorParticipation::new(),
1029            vote_participation: VoteParticipation::new(stake_table, success_threshold, cur_epoch),
1030            drb_upgrade_difficulty,
1031        }
1032    }
1033
1034    /// Get the current view.
1035    pub fn cur_view(&self) -> ViewNumber {
1036        self.cur_view
1037    }
1038
1039    /// Get the current epoch.
1040    pub fn cur_epoch(&self) -> Option<EpochNumber> {
1041        self.cur_epoch
1042    }
1043
1044    /// Get the last decided view.
1045    pub fn last_decided_view(&self) -> ViewNumber {
1046        self.last_decided_view
1047    }
1048
1049    /// Get the locked view.
1050    pub fn locked_view(&self) -> ViewNumber {
1051        self.locked_view
1052    }
1053
1054    /// Get the high QC.
1055    pub fn high_qc(&self) -> &QuorumCertificate2<TYPES> {
1056        &self.high_qc
1057    }
1058
1059    /// Get the transition QC.
1060    pub fn transition_qc(
1061        &self,
1062    ) -> Option<&(
1063        QuorumCertificate2<TYPES>,
1064        NextEpochQuorumCertificate2<TYPES>,
1065    )> {
1066        self.transition_qc.as_ref()
1067    }
1068
1069    ///Update the highest block number
1070    pub fn update_highest_block(&mut self, block_number: u64) {
1071        if block_number > self.highest_block {
1072            self.highest_block = block_number;
1073            return;
1074        }
1075
1076        if is_epoch_transition(block_number, self.epoch_height) {
1077            let new_epoch = epoch_from_block_number(block_number, self.epoch_height);
1078            let high_epoch = epoch_from_block_number(self.highest_block, self.epoch_height);
1079            if new_epoch >= high_epoch {
1080                self.highest_block = block_number;
1081            }
1082        }
1083    }
1084
1085    /// Update the transition QC.
1086    pub fn update_transition_qc(
1087        &mut self,
1088        qc: QuorumCertificate2<TYPES>,
1089        next_epoch_qc: NextEpochQuorumCertificate2<TYPES>,
1090    ) {
1091        if next_epoch_qc.data.leaf_commit != qc.data().leaf_commit {
1092            tracing::error!(
1093                "Next epoch QC for view {} has different leaf commit {:?} to {:?}",
1094                qc.view_number(),
1095                next_epoch_qc.data.leaf_commit,
1096                qc.data().leaf_commit
1097            );
1098            return;
1099        }
1100        if let Some((transition_qc, _)) = &self.transition_qc
1101            && transition_qc.view_number() >= qc.view_number()
1102        {
1103            return;
1104        }
1105        self.transition_qc = Some((qc, next_epoch_qc));
1106    }
1107
1108    /// Get the current light client state certificate
1109    pub fn state_cert(&self) -> Option<&LightClientStateUpdateCertificateV2<TYPES>> {
1110        self.state_cert.as_ref()
1111    }
1112
1113    /// Get the next epoch high QC.
1114    pub fn next_epoch_high_qc(&self) -> Option<&NextEpochQuorumCertificate2<TYPES>> {
1115        self.next_epoch_high_qc.as_ref()
1116    }
1117
1118    /// Get the validated state map.
1119    pub fn validated_state_map(&self) -> &BTreeMap<ViewNumber, View<TYPES>> {
1120        &self.validated_state_map
1121    }
1122
1123    /// Get the saved leaves.
1124    pub fn saved_leaves(&self) -> &CommitmentMap<Leaf2<TYPES>> {
1125        &self.saved_leaves
1126    }
1127
1128    /// Get the saved payloads.
1129    pub fn saved_payloads(&self) -> &BTreeMap<ViewNumber, Arc<PayloadWithMetadata<TYPES>>> {
1130        &self.saved_payloads
1131    }
1132
1133    /// Get the vid shares.
1134    pub fn vid_shares(&self) -> &VidShares<TYPES> {
1135        &self.vid_shares
1136    }
1137
1138    /// Get the saved DA certs.
1139    pub fn saved_da_certs(&self) -> &HashMap<ViewNumber, DaCertificate2<TYPES>> {
1140        &self.saved_da_certs
1141    }
1142
1143    /// Get the map of our recent proposals
1144    pub fn last_proposals(
1145        &self,
1146    ) -> &BTreeMap<ViewNumber, Proposal<TYPES, QuorumProposalWrapper<TYPES>>> {
1147        &self.last_proposals
1148    }
1149
1150    /// Update the current view.
1151    /// # Errors
1152    /// Can return an error when the new view_number is not higher than the existing view number.
1153    pub fn update_view(&mut self, view_number: ViewNumber) -> Result<()> {
1154        ensure!(
1155            view_number > self.cur_view,
1156            debug!("New view isn't newer than the current view.")
1157        );
1158        self.cur_view = view_number;
1159        Ok(())
1160    }
1161
1162    /// Update the validator participation
1163    pub fn update_validator_participation(
1164        &mut self,
1165        key: TYPES::SignatureKey,
1166        epoch: EpochNumber,
1167        proposed: bool,
1168    ) {
1169        self.validator_participation
1170            .update_participation(key, epoch, proposed);
1171    }
1172
1173    /// Update the validator participation epoch
1174    pub fn update_validator_participation_epoch(&mut self, epoch: EpochNumber) {
1175        self.validator_participation
1176            .update_participation_epoch(epoch);
1177    }
1178
1179    /// Get the current proposal participation
1180    pub fn current_proposal_participation(&self) -> HashMap<TYPES::SignatureKey, f64> {
1181        self.validator_participation
1182            .current_proposal_participation()
1183    }
1184
1185    /// Get the proposal participation for a given epoch
1186    pub fn proposal_participation(&self, epoch: EpochNumber) -> HashMap<TYPES::SignatureKey, f64> {
1187        self.validator_participation.proposal_participation(epoch)
1188    }
1189
1190    pub fn update_participation_from_qc(
1191        &mut self,
1192        qc: &QuorumCertificate2<TYPES>,
1193        membership: &EpochMembershipCoordinator<TYPES>,
1194    ) -> Result<()> {
1195        track_decided_qc_participation(
1196            qc,
1197            membership,
1198            &mut self.validator_participation,
1199            &mut self.vote_participation,
1200        )
1201    }
1202
1203    /// Update the vote participation
1204    pub fn update_vote_participation(&mut self, qc: QuorumCertificate2<TYPES>) -> Result<()> {
1205        self.vote_participation.update_participation(qc)
1206    }
1207
1208    /// Update the vote participation epoch
1209    pub fn update_vote_participation_epoch(
1210        &mut self,
1211        stake_table: HSStakeTable<TYPES>,
1212        success_threshold: U256,
1213        epoch: Option<EpochNumber>,
1214    ) -> Result<()> {
1215        self.vote_participation
1216            .update_participation_epoch(stake_table, success_threshold, epoch)
1217    }
1218
1219    /// Get the current vote participation
1220    pub fn current_vote_participation(
1221        &self,
1222    ) -> HashMap<<TYPES::SignatureKey as SignatureKey>::VerificationKeyType, f64> {
1223        self.vote_participation.current_vote_participation()
1224    }
1225
1226    /// Get the previous vote participation
1227    pub fn vote_participation(
1228        &self,
1229        epoch: Option<EpochNumber>,
1230    ) -> HashMap<<TYPES::SignatureKey as SignatureKey>::VerificationKeyType, f64> {
1231        self.vote_participation.vote_participation(epoch)
1232    }
1233
1234    /// Get the parent Leaf Info from a given leaf and our public key.
1235    /// Returns None if we don't have the data in out state
1236    pub async fn parent_leaf_info(
1237        &self,
1238        leaf: &Leaf2<TYPES>,
1239        public_key: &TYPES::SignatureKey,
1240    ) -> Option<LeafInfo<TYPES>> {
1241        let parent_view_number = leaf.justify_qc().view_number();
1242        let parent_epoch = leaf.justify_qc().epoch();
1243        let parent_leaf = self
1244            .saved_leaves
1245            .get(&leaf.justify_qc().data().leaf_commit)?;
1246        let parent_state_and_delta = self.state_and_delta(parent_view_number);
1247        let (Some(state), delta) = parent_state_and_delta else {
1248            return None;
1249        };
1250
1251        let parent_vid = self
1252            .vid_shares()
1253            .get(&parent_view_number)
1254            .and_then(|key_map| key_map.get(public_key))
1255            .and_then(|epoch_map| epoch_map.get(&parent_epoch))
1256            .map(|prop| prop.data.clone());
1257
1258        let state_cert = if parent_leaf.with_epoch
1259            && is_epoch_root(parent_leaf.block_header().block_number(), self.epoch_height)
1260        {
1261            match self.state_cert() {
1262                // Sanity check that the state cert is for the same view as the parent leaf
1263                Some(state_cert)
1264                    if state_cert.light_client_state.view_number == parent_view_number.u64() =>
1265                {
1266                    Some(state_cert.clone())
1267                },
1268                _ => None,
1269            }
1270        } else {
1271            None
1272        };
1273
1274        Some(LeafInfo {
1275            leaf: parent_leaf.clone(),
1276            state,
1277            delta,
1278            vid_share: parent_vid,
1279            state_cert,
1280        })
1281    }
1282
1283    /// Update the current epoch.
1284    /// # Errors
1285    /// Can return an error when the new epoch_number is not higher than the existing epoch number.
1286    pub fn update_epoch(&mut self, epoch_number: EpochNumber) -> Result<()> {
1287        ensure!(
1288            self.cur_epoch.is_none() || Some(epoch_number) > self.cur_epoch,
1289            debug!("New epoch isn't newer than the current epoch.")
1290        );
1291        tracing::trace!(
1292            "Updating epoch from {:?} to {}",
1293            self.cur_epoch,
1294            epoch_number
1295        );
1296        self.cur_epoch = Some(epoch_number);
1297        Ok(())
1298    }
1299
1300    /// Update the last actioned view internally for votes and proposals
1301    ///
1302    /// Returns true if the action is for a newer view than the last action of that type
1303    pub fn update_action(&mut self, action: HotShotAction, view: ViewNumber) -> bool {
1304        let old_view = match action {
1305            HotShotAction::Vote => &mut self.last_actions.voted,
1306            HotShotAction::Propose => &mut self.last_actions.proposed,
1307            HotShotAction::DaPropose => &mut self.last_actions.da_proposed,
1308            HotShotAction::DaVote => {
1309                if view > self.last_actions.da_vote {
1310                    self.last_actions.da_vote = view;
1311                }
1312                // TODO Add logic to prevent double voting.  For now the simple check if
1313                // the last voted view is less than the view we are trying to vote doesn't work
1314                // because the leader of view n + 1 may propose to the DA (and we would vote)
1315                // before the leader of view n.
1316                return true;
1317            },
1318            _ => return true,
1319        };
1320        if view > *old_view {
1321            *old_view = view;
1322            return true;
1323        }
1324        false
1325    }
1326
1327    /// reset last actions to genesis so we can resend events in tests
1328    pub fn reset_actions(&mut self) {
1329        self.last_actions = HotShotActionViews::default();
1330    }
1331
1332    /// Update the last proposal.
1333    ///
1334    /// # Errors
1335    /// Can return an error when the new view_number is not higher than the existing proposed view number.
1336    pub fn update_proposed_view(
1337        &mut self,
1338        proposal: Proposal<TYPES, QuorumProposalWrapper<TYPES>>,
1339    ) -> Result<()> {
1340        ensure!(
1341            proposal.data.view_number()
1342                > self
1343                    .last_proposals
1344                    .last_key_value()
1345                    .map_or(ViewNumber::genesis(), |(k, _)| { *k }),
1346            debug!("New view isn't newer than the previously proposed view.")
1347        );
1348        self.last_proposals
1349            .insert(proposal.data.view_number(), proposal);
1350        Ok(())
1351    }
1352
1353    /// Update the last decided view.
1354    ///
1355    /// # Errors
1356    /// Can return an error when the new view_number is not higher than the existing decided view number.
1357    pub fn update_last_decided_view(&mut self, view_number: ViewNumber) -> Result<()> {
1358        ensure!(
1359            view_number > self.last_decided_view,
1360            debug!("New view isn't newer than the previously decided view.")
1361        );
1362        self.last_decided_view = view_number;
1363        Ok(())
1364    }
1365
1366    /// Update the locked view.
1367    ///
1368    /// # Errors
1369    /// Can return an error when the new view_number is not higher than the existing locked view number.
1370    pub fn update_locked_view(&mut self, view_number: ViewNumber) -> Result<()> {
1371        ensure!(
1372            view_number > self.locked_view,
1373            debug!("New view isn't newer than the previously locked view.")
1374        );
1375        self.locked_view = view_number;
1376        Ok(())
1377    }
1378
1379    /// Update the validated state map with a new view_number/view combo.
1380    ///
1381    /// # Errors
1382    /// Can return an error when the new view contains less information than the existing view
1383    /// with the same view number.
1384    pub fn update_da_view(
1385        &mut self,
1386        view_number: ViewNumber,
1387        epoch: Option<EpochNumber>,
1388        payload_commitment: VidCommitment,
1389    ) -> Result<()> {
1390        let view = View {
1391            view_inner: ViewInner::Da {
1392                payload_commitment,
1393                epoch,
1394            },
1395        };
1396        self.update_validated_state_map(view_number, view)
1397    }
1398
1399    /// Update the validated state map with a new view_number/view combo.
1400    ///
1401    /// # Errors
1402    /// Can return an error when the new view contains less information than the existing view
1403    /// with the same view number.
1404    pub fn update_leaf(
1405        &mut self,
1406        leaf: Leaf2<TYPES>,
1407        state: Arc<TYPES::ValidatedState>,
1408        delta: Option<Arc<<TYPES::ValidatedState as ValidatedState<TYPES>>::Delta>>,
1409    ) -> Result<()> {
1410        let view_number = leaf.view_number();
1411        let epoch =
1412            option_epoch_from_block_number(leaf.with_epoch, leaf.height(), self.epoch_height);
1413        let view = View {
1414            view_inner: ViewInner::Leaf {
1415                leaf: leaf.commit(),
1416                state,
1417                delta,
1418                epoch,
1419            },
1420        };
1421        self.update_validated_state_map(view_number, view)?;
1422        self.update_saved_leaves(leaf);
1423        Ok(())
1424    }
1425
1426    /// Update the validated state map with a new view_number/view combo.
1427    ///
1428    /// # Errors
1429    /// Can return an error when the new view contains less information than the existing view
1430    /// with the same view number.
1431    fn update_validated_state_map(
1432        &mut self,
1433        view_number: ViewNumber,
1434        new_view: View<TYPES>,
1435    ) -> Result<()> {
1436        if let Some(existing_view) = self.validated_state_map().get(&view_number)
1437            && let ViewInner::Leaf {
1438                delta: ref existing_delta,
1439                ..
1440            } = existing_view.view_inner
1441        {
1442            if let ViewInner::Leaf {
1443                delta: ref new_delta,
1444                ..
1445            } = new_view.view_inner
1446            {
1447                ensure!(
1448                    new_delta.is_some() || existing_delta.is_none(),
1449                    debug!(
1450                        "Skipping the state update to not override a `Leaf` view with `Some` \
1451                         state delta."
1452                    )
1453                );
1454            } else {
1455                bail!(
1456                    "Skipping the state update to not override a `Leaf` view with a non-`Leaf` \
1457                     view."
1458                );
1459            }
1460        }
1461        self.validated_state_map.insert(view_number, new_view);
1462        Ok(())
1463    }
1464
1465    /// Update the saved leaves with a new leaf.
1466    fn update_saved_leaves(&mut self, leaf: Leaf2<TYPES>) {
1467        self.saved_leaves.insert(leaf.commit(), leaf);
1468    }
1469
1470    /// Update the saved payloads with a new encoded transaction.
1471    ///
1472    /// # Errors
1473    /// Can return an error when there's an existing payload corresponding to the same view number.
1474    pub fn update_saved_payloads(
1475        &mut self,
1476        view_number: ViewNumber,
1477        payload: Arc<PayloadWithMetadata<TYPES>>,
1478    ) -> Result<()> {
1479        ensure!(
1480            !self.saved_payloads.contains_key(&view_number),
1481            "Payload with the same view already exists."
1482        );
1483        self.saved_payloads.insert(view_number, payload);
1484        Ok(())
1485    }
1486
1487    /// Update the high QC if given a newer one.
1488    /// # Errors
1489    /// Can return an error when the provided high_qc is not newer than the existing entry.
1490    pub fn update_high_qc(&mut self, high_qc: QuorumCertificate2<TYPES>) -> Result<()> {
1491        if self.high_qc == high_qc {
1492            return Ok(());
1493        }
1494        // make sure the we don't update the high QC unless is't a higher view
1495        ensure!(
1496            high_qc.view_number > self.high_qc.view_number,
1497            debug!("High QC with an equal or higher view exists.")
1498        );
1499        tracing::debug!("Updating high QC");
1500        self.high_qc = high_qc;
1501
1502        Ok(())
1503    }
1504
1505    /// Update the next epoch high QC if given a newer one.
1506    /// # Errors
1507    /// Can return an error when the provided high_qc is not newer than the existing entry.
1508    /// # Panics
1509    /// It can't actually panic. If the option is None, we will not call unwrap on it.
1510    pub fn update_next_epoch_high_qc(
1511        &mut self,
1512        high_qc: NextEpochQuorumCertificate2<TYPES>,
1513    ) -> Result<()> {
1514        if self.next_epoch_high_qc.as_ref() == Some(&high_qc) {
1515            return Ok(());
1516        }
1517        if let Some(next_epoch_high_qc) = self.next_epoch_high_qc() {
1518            ensure!(
1519                high_qc.view_number > next_epoch_high_qc.view_number,
1520                debug!("Next epoch high QC with an equal or higher view exists.")
1521            );
1522        }
1523        tracing::debug!("Updating next epoch high QC");
1524        self.next_epoch_high_qc = Some(high_qc);
1525
1526        Ok(())
1527    }
1528
1529    /// Resets high qc and next epoch qc to the provided transition qc.
1530    /// # Errors
1531    /// Can return an error when the provided high_qc is not newer than the existing entry.
1532    pub fn reset_high_qc(
1533        &mut self,
1534        high_qc: QuorumCertificate2<TYPES>,
1535        next_epoch_qc: NextEpochQuorumCertificate2<TYPES>,
1536    ) -> Result<()> {
1537        ensure!(
1538            high_qc.data.leaf_commit == next_epoch_qc.data.leaf_commit,
1539            error!("High QC's and next epoch QC's leaf commits do not match.")
1540        );
1541        if self.high_qc == high_qc {
1542            return Ok(());
1543        }
1544        let same_epoch = high_qc.data.block_number.is_some_and(|bn| {
1545            let current_qc = self.high_qc();
1546            let Some(high_bn) = current_qc.data.block_number else {
1547                return false;
1548            };
1549            epoch_from_block_number(bn + 1, self.epoch_height)
1550                == epoch_from_block_number(high_bn + 1, self.epoch_height)
1551        });
1552        ensure!(
1553            high_qc
1554                .data
1555                .block_number
1556                .is_some_and(|bn| is_transition_block(bn, self.epoch_height))
1557                && same_epoch,
1558            error!("Provided QC is not a transition QC.")
1559        );
1560        tracing::debug!("Resetting high QC and next epoch high QC");
1561        self.high_qc = high_qc;
1562        self.next_epoch_high_qc = Some(next_epoch_qc);
1563
1564        Ok(())
1565    }
1566
1567    /// Update the light client state update certificate if given a newer one.
1568    /// # Errors
1569    /// Can return an error when the provided state_cert is not newer than the existing entry.
1570    pub fn update_state_cert(
1571        &mut self,
1572        state_cert: LightClientStateUpdateCertificateV2<TYPES>,
1573    ) -> Result<()> {
1574        if let Some(existing_state_cert) = &self.state_cert {
1575            ensure!(
1576                state_cert.epoch > existing_state_cert.epoch,
1577                debug!(
1578                    "Light client state update certification with an equal or higher epoch exists."
1579                )
1580            );
1581        }
1582        tracing::debug!("Updating light client state update certification");
1583        self.state_cert = Some(state_cert);
1584
1585        Ok(())
1586    }
1587
1588    /// Add a new entry to the vid_shares map.
1589    pub fn update_vid_shares(
1590        &mut self,
1591        view_number: ViewNumber,
1592        disperse: Proposal<TYPES, VidDisperseShare<TYPES>>,
1593    ) {
1594        self.vid_shares
1595            .entry(view_number)
1596            .or_default()
1597            .entry(disperse.data.recipient_key().clone())
1598            .or_default()
1599            .insert(disperse.data.target_epoch(), disperse);
1600    }
1601
1602    /// Add a new entry to the da_certs map.
1603    pub fn update_saved_da_certs(&mut self, view_number: ViewNumber, cert: DaCertificate2<TYPES>) {
1604        self.saved_da_certs.insert(view_number, cert);
1605    }
1606
1607    /// gather information from the parent chain of leaves
1608    /// # Errors
1609    /// If the leaf or its ancestors are not found in storage
1610    pub fn visit_leaf_ancestors<F>(
1611        &self,
1612        start_from: ViewNumber,
1613        terminator: Terminator<ViewNumber>,
1614        ok_when_finished: bool,
1615        mut f: F,
1616    ) -> std::result::Result<(), HotShotError<TYPES>>
1617    where
1618        F: FnMut(
1619            &Leaf2<TYPES>,
1620            Arc<<TYPES as NodeType>::ValidatedState>,
1621            Option<Arc<<<TYPES as NodeType>::ValidatedState as ValidatedState<TYPES>>::Delta>>,
1622        ) -> bool,
1623    {
1624        let mut next_leaf = if let Some(view) = self.validated_state_map.get(&start_from) {
1625            view.leaf_commitment().ok_or_else(|| {
1626                HotShotError::InvalidState(format!(
1627                    "Visited failed view {start_from} leaf. Expected successful leaf"
1628                ))
1629            })?
1630        } else {
1631            return Err(HotShotError::InvalidState(format!(
1632                "View {start_from} leaf does not exist in state map "
1633            )));
1634        };
1635
1636        while let Some(leaf) = self.saved_leaves.get(&next_leaf) {
1637            let view = leaf.view_number();
1638            if let (Some(state), delta) = self.state_and_delta(view) {
1639                if let Terminator::Exclusive(stop_before) = terminator
1640                    && stop_before == view
1641                {
1642                    if ok_when_finished {
1643                        return Ok(());
1644                    }
1645                    break;
1646                }
1647                next_leaf = leaf.parent_commitment();
1648                if !f(leaf, state, delta) {
1649                    return Ok(());
1650                }
1651                if let Terminator::Inclusive(stop_after) = terminator
1652                    && stop_after == view
1653                {
1654                    if ok_when_finished {
1655                        return Ok(());
1656                    }
1657                    break;
1658                }
1659            } else {
1660                return Err(HotShotError::InvalidState(format!(
1661                    "View {view} state does not exist in state map"
1662                )));
1663            }
1664        }
1665        Err(HotShotError::MissingLeaf(next_leaf))
1666    }
1667
1668    /// Garbage collects based on state change right now, this removes from both the
1669    /// `saved_payloads` and `validated_state_map` fields of `Consensus`.
1670    /// # Panics
1671    /// On inconsistent stored entries
1672    pub fn collect_garbage(&mut self, old_anchor_view: ViewNumber, new_anchor_view: ViewNumber) {
1673        // Nothing to collect
1674        if new_anchor_view <= old_anchor_view {
1675            return;
1676        }
1677        let gc_view = ViewNumber::new(new_anchor_view.saturating_sub(1));
1678        // state check
1679        let anchor_entry = self
1680            .validated_state_map
1681            .iter()
1682            .next()
1683            .expect("INCONSISTENT STATE: anchor leaf not in state map!");
1684        if **anchor_entry.0 != old_anchor_view.saturating_sub(1) {
1685            tracing::info!(
1686                "Something about GC has failed. Older leaf exists than the previous anchor leaf."
1687            );
1688        }
1689        // perform gc
1690        self.saved_da_certs
1691            .retain(|view_number, _| *view_number >= old_anchor_view);
1692        self.validated_state_map
1693            .range(..gc_view)
1694            .filter_map(|(_view_number, view)| view.leaf_commitment())
1695            .for_each(|leaf| {
1696                self.saved_leaves.remove(&leaf);
1697            });
1698        self.validated_state_map = self.validated_state_map.split_off(&gc_view);
1699        self.saved_payloads = self.saved_payloads.split_off(&gc_view);
1700        self.vid_shares = self.vid_shares.split_off(&gc_view);
1701        self.last_proposals = self.last_proposals.split_off(&gc_view);
1702    }
1703
1704    /// Gets the last decided leaf.
1705    ///
1706    /// # Panics
1707    /// if the last decided view's leaf does not exist in the state map or saved leaves, which
1708    /// should never happen.
1709    #[must_use]
1710    pub fn decided_leaf(&self) -> Leaf2<TYPES> {
1711        let decided_view_num = self.last_decided_view;
1712        let view = self.validated_state_map.get(&decided_view_num).unwrap();
1713        let leaf = view
1714            .leaf_commitment()
1715            .expect("Decided leaf not found! Consensus internally inconsistent");
1716        self.saved_leaves.get(&leaf).unwrap().clone()
1717    }
1718
1719    pub fn undecided_leaves(&self) -> Vec<Leaf2<TYPES>> {
1720        self.saved_leaves.values().cloned().collect::<Vec<_>>()
1721    }
1722
1723    /// Gets the validated state with the given view number, if in the state map.
1724    #[must_use]
1725    pub fn state(&self, view_number: ViewNumber) -> Option<&Arc<TYPES::ValidatedState>> {
1726        match self.validated_state_map.get(&view_number) {
1727            Some(view) => view.state(),
1728            None => None,
1729        }
1730    }
1731
1732    /// Gets the validated state and state delta with the given view number, if in the state map.
1733    #[must_use]
1734    pub fn state_and_delta(&self, view_number: ViewNumber) -> StateAndDelta<TYPES> {
1735        match self.validated_state_map.get(&view_number) {
1736            Some(view) => view.state_and_delta(),
1737            None => (None, None),
1738        }
1739    }
1740
1741    /// Gets the last decided validated state.
1742    ///
1743    /// # Panics
1744    /// If the last decided view's state does not exist in the state map, which should never
1745    /// happen.
1746    #[must_use]
1747    pub fn decided_state(&self) -> Arc<TYPES::ValidatedState> {
1748        let decided_view_num = self.last_decided_view;
1749        self.state_and_delta(decided_view_num)
1750            .0
1751            .expect("Decided state not found! Consensus internally inconsistent")
1752    }
1753
1754    /// Associated helper function:
1755    /// Takes `LockedConsensusState` which will be updated; locks it for read and write accordingly.
1756    /// Calculates `VidDisperse` based on the view, the txns and the membership,
1757    /// and updates `vid_shares` map with the signed `VidDisperseShare` proposals.
1758    /// Returned `Option` indicates whether the update has actually happened or not.
1759    #[instrument(skip_all, target = "Consensus", fields(view = *view))]
1760    pub async fn calculate_and_update_vid(
1761        consensus: OuterConsensus<TYPES>,
1762        view: ViewNumber,
1763        target_epoch: Option<EpochNumber>,
1764        membership_coordinator: EpochMembershipCoordinator<TYPES>,
1765        private_key: &<TYPES::SignatureKey as SignatureKey>::PrivateKey,
1766        upgrade_lock: &UpgradeLock<TYPES>,
1767    ) -> Option<()> {
1768        let payload_with_metadata = Arc::clone(consensus.read().await.saved_payloads().get(&view)?);
1769        let epoch = consensus
1770            .read()
1771            .await
1772            .validated_state_map()
1773            .get(&view)?
1774            .view_inner
1775            .epoch()?;
1776
1777        let VidDisperseAndDuration {
1778            disperse: vid,
1779            duration: disperse_duration,
1780        } = VidDisperse::calculate_vid_disperse(
1781            &payload_with_metadata.payload,
1782            &membership_coordinator,
1783            view,
1784            target_epoch,
1785            epoch,
1786            &payload_with_metadata.metadata,
1787            upgrade_lock,
1788        )
1789        .await
1790        .ok()?;
1791
1792        let mut consensus_writer = consensus.write().await;
1793        consensus_writer
1794            .metrics
1795            .vid_disperse_duration
1796            .add_point(disperse_duration.as_secs_f64());
1797        for share in vid.to_shares() {
1798            if let Some(prop) = share.to_proposal(private_key) {
1799                consensus_writer.update_vid_shares(view, prop);
1800            }
1801        }
1802
1803        Some(())
1804    }
1805    /// Returns true if a given leaf is for the epoch transition
1806    pub fn is_epoch_transition(&self, leaf_commit: LeafCommitment<TYPES>) -> bool {
1807        let Some(leaf) = self.saved_leaves.get(&leaf_commit) else {
1808            tracing::trace!("We don't have a leaf corresponding to the leaf commit");
1809            return false;
1810        };
1811        let block_height = leaf.height();
1812        is_epoch_transition(block_height, self.epoch_height)
1813    }
1814
1815    /// Returns true if our high QC is for one of the epoch transition blocks
1816    pub fn is_high_qc_for_epoch_transition(&self) -> bool {
1817        let Some(block_height) = self.high_qc().data.block_number else {
1818            return false;
1819        };
1820        is_epoch_transition(block_height, self.epoch_height)
1821    }
1822
1823    /// Returns true if the `parent_leaf` formed an eQC for the previous epoch to the `proposed_leaf`
1824    pub fn check_eqc(&self, proposed_leaf: &Leaf2<TYPES>, parent_leaf: &Leaf2<TYPES>) -> bool {
1825        if parent_leaf.view_number() == ViewNumber::genesis() {
1826            return true;
1827        }
1828        let new_epoch = epoch_from_block_number(proposed_leaf.height(), self.epoch_height);
1829        let old_epoch = epoch_from_block_number(parent_leaf.height(), self.epoch_height);
1830
1831        new_epoch - 1 == old_epoch && is_last_block(parent_leaf.height(), self.epoch_height)
1832    }
1833}
1834
1835/// Alias for the block payload commitment and the associated metadata. The primary data
1836/// needed in order to submit a proposal.
1837#[derive(Eq, PartialEq, Debug, Clone)]
1838pub struct CommitmentAndMetadata<TYPES: NodeType> {
1839    /// Vid Commitment
1840    pub commitment: VidCommitment,
1841    /// Builder Commitment
1842    pub builder_commitment: BuilderCommitment,
1843    /// Metadata for the block payload
1844    pub metadata: <TYPES::BlockPayload as BlockPayload<TYPES>>::Metadata,
1845    /// Builder fee data
1846    pub fees: Vec1<BuilderFee<TYPES>>,
1847    /// View number this block is for
1848    pub block_view: ViewNumber,
1849}
hotshot_types/consensus.rs

hotshot_types/
consensus.rs
