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::collections::{BTreeMap, HashMap, HashSet};
10
11use alloy_primitives::U256;
12use committable::Commitment;
13use hotshot_utils::anytrace::*;
14
15pub use crate::utils::{View, ViewInner};
16use crate::{
17    constants::EPOCH_PARTICIPATION_HISTORY,
18    data::{EpochNumber, Leaf2, VidDisperseShare, ViewNumber},
19    epoch_membership::EpochMembershipCoordinator,
20    message::Proposal,
21    simple_certificate::QuorumCertificate2,
22    simple_vote::HasEpoch,
23    stake_table::{HSStakeTable, StakeTableEntries},
24    traits::{
25        BlockPayload,
26        metrics::{Counter, Gauge, Histogram, Metrics, NoMetrics},
27        node_implementation::NodeType,
28        signature_key::{SignatureKey, StakeTableEntryType},
29    },
30    vote::{Certificate, HasViewNumber},
31};
32
33/// A type alias for `HashMap<Commitment<T>, T>`
34pub type CommitmentMap<T> = HashMap<Commitment<T>, T>;
35
36/// A type alias for `BTreeMap<T::Time, HashMap<T::SignatureKey, BTreeMap<T::Epoch, Proposal<T, VidDisperseShare<T>>>>>`
37pub type VidShares<TYPES> = BTreeMap<
38    ViewNumber,
39    HashMap<
40        <TYPES as NodeType>::SignatureKey,
41        BTreeMap<Option<EpochNumber>, Proposal<TYPES, VidDisperseShare<TYPES>>>,
42    >,
43>;
44
45type ValidatorParticipationMap<TYPES> = HashMap<<TYPES as NodeType>::SignatureKey, (u64, u64)>;
46
47/// Tracks proposal and vote participation per validator across epochs.
48///
49/// The new-protocol coordinator owns one of these and serves the node
50/// participation API from it.
51pub struct ParticipationTracker<TYPES: NodeType> {
52    validator: ValidatorParticipation<TYPES>,
53    vote: VoteParticipation<TYPES>,
54}
55
56impl<TYPES: NodeType> Default for ParticipationTracker<TYPES> {
57    fn default() -> Self {
58        let (stake_table, success_threshold) = VoteParticipation::unresolved_stake_table();
59        Self {
60            validator: ValidatorParticipation::new(),
61            vote: VoteParticipation::new(stake_table, success_threshold, EpochNumber::genesis()),
62        }
63    }
64}
65
66impl<TYPES: NodeType> ParticipationTracker<TYPES> {
67    pub fn new(membership: &EpochMembershipCoordinator<TYPES>, epoch: EpochNumber) -> Self {
68        let (stake_table, success_threshold) = resolve_participation_stake_table(membership, epoch);
69        Self {
70            validator: ValidatorParticipation::new_in_epoch(epoch),
71            vote: VoteParticipation::new(stake_table, success_threshold, epoch),
72        }
73    }
74
75    pub fn leader_proposed(&mut self, leader: TYPES::SignatureKey, epoch: EpochNumber) {
76        self.validator.update_participation(leader, epoch, true);
77    }
78
79    pub fn leader_missed(&mut self, leader: TYPES::SignatureKey, epoch: EpochNumber) {
80        self.validator.update_participation(leader, epoch, false);
81    }
82
83    pub fn on_view_changed(&mut self, epoch: EpochNumber) {
84        self.validator.update_participation_epoch(epoch);
85    }
86
87    pub fn on_leaf_decided(
88        &mut self,
89        leaf: &Leaf2<TYPES>,
90        membership: &EpochMembershipCoordinator<TYPES>,
91    ) {
92        if let Err(err) = track_decided_qc_participation(
93            leaf.justify_qc(),
94            membership,
95            &mut self.validator,
96            &mut self.vote,
97        ) {
98            tracing::warn!(%err, "failed to update vote participation epoch");
99        }
100    }
101
102    pub fn current_proposal_participation(&self) -> HashMap<TYPES::SignatureKey, f64> {
103        self.validator.current_proposal_participation()
104    }
105
106    pub fn proposal_participation(&self, epoch: EpochNumber) -> HashMap<TYPES::SignatureKey, f64> {
107        self.validator.proposal_participation(epoch)
108    }
109
110    pub fn current_vote_participation(
111        &self,
112    ) -> HashMap<<TYPES::SignatureKey as SignatureKey>::VerificationKeyType, f64> {
113        self.vote.current_vote_participation()
114    }
115
116    pub fn vote_participation(
117        &self,
118        epoch: EpochNumber,
119    ) -> HashMap<<TYPES::SignatureKey as SignatureKey>::VerificationKeyType, f64> {
120        self.vote.vote_participation(epoch)
121    }
122}
123
124#[derive(Debug, Clone)]
125struct ValidatorParticipation<TYPES: NodeType> {
126    epoch: EpochNumber,
127    /// Current epoch participation by key maps key -> (num leader, num times proposed)
128    current_epoch_participation: ValidatorParticipationMap<TYPES>,
129
130    /// Previous epochs participation ordered by epoch number, maps epoch -> key -> (num leader, num times proposed)
131    previous_epoch_participation: BTreeMap<EpochNumber, ValidatorParticipationMap<TYPES>>,
132}
133
134impl<TYPES: NodeType> ValidatorParticipation<TYPES> {
135    fn new() -> Self {
136        Self::new_in_epoch(EpochNumber::genesis())
137    }
138
139    fn new_in_epoch(epoch: EpochNumber) -> Self {
140        Self {
141            epoch,
142            current_epoch_participation: HashMap::new(),
143            previous_epoch_participation: BTreeMap::new(),
144        }
145    }
146
147    fn update_participation(
148        &mut self,
149        key: TYPES::SignatureKey,
150        epoch: EpochNumber,
151        proposed: bool,
152    ) {
153        let participation = match epoch.cmp(&self.epoch) {
154            std::cmp::Ordering::Greater => {
155                self.update_participation_epoch(epoch);
156                &mut self.current_epoch_participation
157            },
158            std::cmp::Ordering::Equal => &mut self.current_epoch_participation,
159            std::cmp::Ordering::Less => {
160                let Some(archived) = self.previous_epoch_participation.get_mut(&epoch) else {
161                    return;
162                };
163                archived
164            },
165        };
166        let entry = participation.entry(key).or_insert((0, 0));
167        if proposed {
168            entry.1 += 1;
169        }
170        entry.0 += 1;
171    }
172
173    fn update_participation_epoch(&mut self, epoch: EpochNumber) {
174        if epoch <= self.epoch {
175            return;
176        }
177        self.previous_epoch_participation
178            .insert(self.epoch, self.current_epoch_participation.clone());
179
180        self.previous_epoch_participation =
181            self.previous_epoch_participation
182                .split_off(&EpochNumber::new(
183                    self.epoch.saturating_sub(EPOCH_PARTICIPATION_HISTORY),
184                ));
185
186        self.epoch = epoch;
187        self.current_epoch_participation = HashMap::new();
188    }
189
190    fn current_proposal_participation(&self) -> HashMap<TYPES::SignatureKey, f64> {
191        self.current_epoch_participation
192            .iter()
193            .map(|(key, (leader, proposed))| {
194                (
195                    key.clone(),
196                    if *leader == 0 {
197                        0.0
198                    } else {
199                        *proposed as f64 / *leader as f64
200                    },
201                )
202            })
203            .collect()
204    }
205    fn proposal_participation(&self, epoch: EpochNumber) -> HashMap<TYPES::SignatureKey, f64> {
206        let tracked_participation = if epoch == self.epoch {
207            self.current_epoch_participation.clone()
208        } else {
209            self.previous_epoch_participation
210                .get(&epoch)
211                .unwrap_or(&HashMap::new())
212                .clone()
213        };
214
215        tracked_participation
216            .iter()
217            .map(|(key, (leader, proposed))| {
218                (
219                    key.clone(),
220                    if *leader == 0 {
221                        0.0
222                    } else {
223                        *proposed as f64 / *leader as f64
224                    },
225                )
226            })
227            .collect()
228    }
229
230    fn current_epoch(&self) -> EpochNumber {
231        self.epoch
232    }
233}
234
235type VoteParticipationMap<TYPES> = (
236    HashMap<<<TYPES as NodeType>::SignatureKey as SignatureKey>::VerificationKeyType, u64>,
237    u64,
238);
239
240#[derive(Clone, Debug)]
241struct VoteParticipation<TYPES: NodeType> {
242    /// Current epoch
243    epoch: EpochNumber,
244
245    /// Current stake_table
246    stake_table: HSStakeTable<TYPES>,
247
248    /// Success threshold
249    success_threshold: U256,
250
251    /// Set of views in the current epoch
252    view_set: HashSet<ViewNumber>,
253
254    /// Number of views in the current epoch
255    current_epoch_num_views: u64,
256
257    /// Current epoch participation by key maps key -> num times voted
258    current_epoch_participation:
259        HashMap<<TYPES::SignatureKey as SignatureKey>::VerificationKeyType, u64>,
260
261    /// Previous epochs participation ordered by epoch number, maps epoch -> key -> num times voted
262    previous_epoch_participation: BTreeMap<EpochNumber, VoteParticipationMap<TYPES>>,
263}
264
265impl<TYPES: NodeType> VoteParticipation<TYPES> {
266    fn new(stake_table: HSStakeTable<TYPES>, success_threshold: U256, epoch: EpochNumber) -> Self {
267        let current_epoch_participation: HashMap<_, _> = stake_table
268            .iter()
269            .map({
270                |peer_config| {
271                    (
272                        peer_config
273                            .stake_table_entry
274                            .public_key()
275                            .to_verification_key(),
276                        0u64,
277                    )
278                }
279            })
280            .collect();
281        Self {
282            epoch,
283            stake_table,
284            success_threshold,
285            view_set: HashSet::new(),
286            current_epoch_num_views: 0u64,
287            current_epoch_participation,
288            previous_epoch_participation: BTreeMap::new(),
289        }
290    }
291
292    fn unresolved_stake_table() -> (HSStakeTable<TYPES>, U256) {
293        (HSStakeTable::default(), U256::MAX)
294    }
295
296    fn stake_table_unresolved(&self) -> bool {
297        self.stake_table.is_empty()
298    }
299
300    fn reseed_stake_table(&mut self, stake_table: HSStakeTable<TYPES>, threshold: U256) {
301        self.current_epoch_participation = stake_table
302            .iter()
303            .map(|peer_config| {
304                (
305                    peer_config
306                        .stake_table_entry
307                        .public_key()
308                        .to_verification_key(),
309                    0u64,
310                )
311            })
312            .collect();
313        self.stake_table = stake_table;
314        self.success_threshold = threshold;
315    }
316
317    fn update_participation(&mut self, qc: QuorumCertificate2<TYPES>) -> Result<()> {
318        ensure!(
319            qc.epoch() == Some(self.epoch),
320            info!(
321                "Incorrect epoch while updating vote participation, current epoch: {:?}, QC epoch \
322                 {:?}",
323                self.epoch,
324                qc.epoch()
325            )
326        );
327        ensure!(
328            !self.view_set.contains(&qc.view_number()),
329            info!(
330                "Participation for view {} already updated",
331                qc.view_number()
332            )
333        );
334        let signers = qc
335            .signers(
336                &StakeTableEntries::<TYPES>::from(self.stake_table.clone()).0,
337                self.success_threshold,
338            )
339            .context(|e| warn!("Tracing signers: {e}"))?;
340        for vk in signers {
341            let Some(votes) = self.current_epoch_participation.get_mut(&vk) else {
342                bail!(warn!(
343                    "Trying to update vote participation for unknown key: {:?}",
344                    vk
345                ));
346            };
347            *votes += 1;
348        }
349        self.view_set.insert(qc.view_number());
350        self.current_epoch_num_views += 1;
351        Ok(())
352    }
353
354    fn update_participation_epoch(
355        &mut self,
356        stake_table: HSStakeTable<TYPES>,
357        success_threshold: U256,
358        epoch: EpochNumber,
359    ) -> Result<()> {
360        ensure!(
361            epoch >= self.epoch,
362            warn!(
363                "New epoch less than current epoch while updating vote participation epoch, \
364                 current epoch: {:?}, new epoch {:?}",
365                self.epoch, epoch
366            )
367        );
368        // Same epoch, do nothing
369        if epoch == self.epoch {
370            return Ok(());
371        }
372
373        self.previous_epoch_participation.insert(
374            self.epoch,
375            (
376                self.current_epoch_participation.clone(),
377                self.current_epoch_num_views,
378            ),
379        );
380
381        self.previous_epoch_participation =
382            self.previous_epoch_participation
383                .split_off(&EpochNumber::new(
384                    self.epoch.saturating_sub(EPOCH_PARTICIPATION_HISTORY),
385                ));
386
387        self.epoch = epoch;
388        self.current_epoch_num_views = 0;
389        self.view_set = HashSet::new();
390        let current_epoch_participation: HashMap<_, _> = stake_table
391            .iter()
392            .map({
393                |peer_config| {
394                    (
395                        peer_config
396                            .stake_table_entry
397                            .public_key()
398                            .to_verification_key(),
399                        0u64,
400                    )
401                }
402            })
403            .collect();
404        self.current_epoch_participation = current_epoch_participation;
405        self.stake_table = stake_table;
406        self.success_threshold = success_threshold;
407        Ok(())
408    }
409
410    fn current_vote_participation(
411        &self,
412    ) -> HashMap<<TYPES::SignatureKey as SignatureKey>::VerificationKeyType, f64> {
413        self.current_epoch_participation
414            .iter()
415            .map(|(key, votes)| {
416                (
417                    key.clone(),
418                    Self::calculate_ratio(votes, self.current_epoch_num_views),
419                )
420            })
421            .collect()
422    }
423    fn vote_participation(
424        &self,
425        epoch: EpochNumber,
426    ) -> HashMap<<TYPES::SignatureKey as SignatureKey>::VerificationKeyType, f64> {
427        let (participation, num_views) = if epoch == self.epoch {
428            (
429                self.current_epoch_participation.clone(),
430                self.current_epoch_num_views,
431            )
432        } else {
433            self.previous_epoch_participation
434                .get(&epoch)
435                .unwrap_or(&(HashMap::new(), 0))
436                .clone()
437        };
438
439        if num_views == 0 {
440            return HashMap::new();
441        }
442
443        participation
444            .iter()
445            .map(|(key, votes)| (key.clone(), Self::calculate_ratio(votes, num_views)))
446            .collect()
447    }
448
449    fn calculate_ratio(num_votes: &u64, total_views: u64) -> f64 {
450        if total_views == 0 {
451            0.0
452        } else {
453            *num_votes as f64 / total_views as f64
454        }
455    }
456
457    fn current_epoch(&self) -> EpochNumber {
458        self.epoch
459    }
460}
461
462fn resolve_participation_stake_table<TYPES: NodeType>(
463    membership: &EpochMembershipCoordinator<TYPES>,
464    epoch: EpochNumber,
465) -> (HSStakeTable<TYPES>, U256) {
466    match membership.stake_table_for_epoch(Some(epoch)) {
467        Ok(m) => (
468            HSStakeTable::from_iter(m.stake_table()),
469            m.success_threshold(),
470        ),
471        Err(err) => {
472            tracing::warn!(?epoch, %err, "no stake table for participation tracking");
473            VoteParticipation::unresolved_stake_table()
474        },
475    }
476}
477
478fn track_decided_qc_participation<TYPES: NodeType>(
479    qc: &QuorumCertificate2<TYPES>,
480    membership: &EpochMembershipCoordinator<TYPES>,
481    validator: &mut ValidatorParticipation<TYPES>,
482    vote: &mut VoteParticipation<TYPES>,
483) -> Result<()> {
484    let Some(qc_epoch) = qc.epoch() else {
485        bail!(warn!(
486            "Decided QC for view {} has no epoch",
487            qc.view_number()
488        ));
489    };
490    if qc_epoch > validator.current_epoch() {
491        validator.update_participation_epoch(qc_epoch);
492    }
493    if qc_epoch > vote.current_epoch() {
494        let (stake_table, success_threshold) =
495            resolve_participation_stake_table(membership, qc_epoch);
496        vote.update_participation_epoch(stake_table, success_threshold, qc_epoch)
497            .context(warn!("Updating vote participation"))?;
498    } else if qc_epoch == vote.current_epoch()
499        && vote.stake_table_unresolved()
500        && let Ok(m) = membership.stake_table_for_epoch(Some(qc_epoch))
501    {
502        vote.reseed_stake_table(
503            HSStakeTable::from_iter(m.stake_table()),
504            m.success_threshold(),
505        );
506    }
507    if let Err(e) = vote.update_participation(qc.clone()) {
508        tracing::warn!("Failed to update vote participation: {e}");
509    }
510    Ok(())
511}
512
513/// This struct holds a payload and its metadata
514#[derive(Debug, Clone, Hash, Eq, PartialEq)]
515pub struct PayloadWithMetadata<TYPES: NodeType> {
516    pub payload: TYPES::BlockPayload,
517    pub metadata: <TYPES::BlockPayload as BlockPayload<TYPES>>::Metadata,
518}
519
520/// Contains several `ConsensusMetrics` that we're interested in from the consensus interfaces
521#[derive(Clone, Debug)]
522pub struct ConsensusMetricsValue {
523    /// The number of last synced block height
524    pub last_synced_block_height: Box<dyn Gauge>,
525    /// The number of last decided view
526    pub last_decided_view: Box<dyn Gauge>,
527    /// The number of the last voted view
528    pub last_voted_view: Box<dyn Gauge>,
529    /// Number of timestamp for the last decided time
530    pub last_decided_time: Box<dyn Gauge>,
531    /// The current view
532    pub current_view: Box<dyn Gauge>,
533    /// Number of views that are in-flight since the last decided view
534    pub number_of_views_since_last_decide: Box<dyn Gauge>,
535    /// Number of views that are in-flight since the last anchor view
536    pub number_of_views_per_decide_event: Box<dyn Histogram>,
537    /// Duration of views as leader
538    pub view_duration_as_leader: Box<dyn Histogram>,
539    /// Number of invalid QCs we've seen since the last commit.
540    pub invalid_qc: Box<dyn Gauge>,
541    /// Number of outstanding transactions
542    pub outstanding_transactions: Box<dyn Gauge>,
543    /// Memory size in bytes of the serialized transactions still outstanding
544    pub outstanding_transactions_memory_size: Box<dyn Gauge>,
545    /// Number of views that timed out
546    pub number_of_timeouts: Box<dyn Counter>,
547    /// Number of views that timed out as leader
548    pub number_of_timeouts_as_leader: Box<dyn Counter>,
549    /// The number of empty blocks that have been proposed
550    pub number_of_empty_blocks_proposed: Box<dyn Counter>,
551    /// Number of events in the hotshot event queue
552    pub internal_event_queue_len: Box<dyn Gauge>,
553    /// Time from proposal creation to decide time
554    pub proposal_to_decide_time: Box<dyn Histogram>,
555    /// Time from proposal received to proposal creation
556    pub previous_proposal_to_proposal_time: Box<dyn Histogram>,
557    /// Finalized bytes per view
558    pub finalized_bytes: Box<dyn Histogram>,
559    /// The duration of the validate and apply header
560    pub validate_and_apply_header_duration: Box<dyn Histogram>,
561    /// The duration of update leaf
562    pub update_leaf_duration: Box<dyn Histogram>,
563    /// The time a leader took for a built block's commitments, mostly erasure-coding the payload
564    /// into the shares it later disperses
565    pub block_commitments_duration: Box<dyn Histogram>,
566    /// The time a leader took to disperse a block: signing and sending the shares, plus
567    /// erasure-coding the payload when the block builder had not
568    pub vid_disperse_duration: Box<dyn Histogram>,
569}
570
571impl ConsensusMetricsValue {
572    /// Create a new instance of this [`ConsensusMetricsValue`] struct, setting all the counters and gauges
573    #[must_use]
574    pub fn new(metrics: &dyn Metrics) -> Self {
575        Self {
576            last_synced_block_height: metrics
577                .create_gauge(String::from("last_synced_block_height"), None),
578            last_decided_view: metrics.create_gauge(String::from("last_decided_view"), None),
579            last_voted_view: metrics.create_gauge(String::from("last_voted_view"), None),
580            last_decided_time: metrics.create_gauge(String::from("last_decided_time"), None),
581            current_view: metrics.create_gauge(String::from("current_view"), None),
582            number_of_views_since_last_decide: metrics
583                .create_gauge(String::from("number_of_views_since_last_decide"), None),
584            number_of_views_per_decide_event: metrics
585                .create_histogram(String::from("number_of_views_per_decide_event"), None),
586            view_duration_as_leader: metrics
587                .create_histogram(String::from("view_duration_as_leader"), None),
588            invalid_qc: metrics.create_gauge(String::from("invalid_qc"), None),
589            outstanding_transactions: metrics
590                .create_gauge(String::from("outstanding_transactions"), None),
591            outstanding_transactions_memory_size: metrics
592                .create_gauge(String::from("outstanding_transactions_memory_size"), None),
593            number_of_timeouts: metrics.create_counter(String::from("number_of_timeouts"), None),
594            number_of_timeouts_as_leader: metrics
595                .create_counter(String::from("number_of_timeouts_as_leader"), None),
596            number_of_empty_blocks_proposed: metrics
597                .create_counter(String::from("number_of_empty_blocks_proposed"), None),
598            internal_event_queue_len: metrics
599                .create_gauge(String::from("internal_event_queue_len"), None),
600            proposal_to_decide_time: metrics
601                .create_histogram(String::from("proposal_to_decide_time"), None),
602            previous_proposal_to_proposal_time: metrics
603                .create_histogram(String::from("previous_proposal_to_proposal_time"), None),
604            finalized_bytes: metrics.create_histogram(String::from("finalized_bytes"), None),
605            validate_and_apply_header_duration: metrics.create_histogram(
606                String::from("validate_and_apply_header_duration"),
607                Some("seconds".to_string()),
608            ),
609            update_leaf_duration: metrics.create_histogram(
610                String::from("update_leaf_duration"),
611                Some("seconds".to_string()),
612            ),
613            block_commitments_duration: metrics.create_histogram(
614                String::from("block_commitments_duration"),
615                Some("seconds".to_string()),
616            ),
617            vid_disperse_duration: metrics.create_histogram(
618                String::from("vid_disperse_duration"),
619                Some("seconds".to_string()),
620            ),
621        }
622    }
623}
624
625impl Default for ConsensusMetricsValue {
626    fn default() -> Self {
627        Self::new(&*NoMetrics::boxed())
628    }
629}
hotshot_types/consensus.rs

hotshot_types/
consensus.rs
