arc-malachitebft-engine 0.7.0-pre

Implementation of the Malachite BFT consensus engine
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142

use core::fmt;
use std::io;
use std::sync::Arc;

use derive_where::derive_where;
use tokio::sync::broadcast;

use malachitebft_core_consensus::{
    Error as ConsensusError, LocallyProposedValue, MisbehaviorEvidence, ProposedValue, Role,
    SignedConsensusMsg, WalEntry,
};
use malachitebft_core_types::{
    CommitCertificate, Context, PolkaCertificate, Round, RoundCertificate, SignedVote, ValueOrigin,
};

pub type RxEvent<Ctx> = broadcast::Receiver<Event<Ctx>>;

#[derive_where(Clone)]
pub struct TxEvent<Ctx: Context> {
    tx: broadcast::Sender<Event<Ctx>>,
}

impl<Ctx: Context> TxEvent<Ctx> {
    pub fn new() -> Self {
        let (tx, _) = broadcast::channel(128);
        Self { tx }
    }

    pub fn subscribe(&self) -> broadcast::Receiver<Event<Ctx>> {
        self.tx.subscribe()
    }

    pub fn send(&self, event: impl FnOnce() -> Event<Ctx>) {
        if self.tx.receiver_count() > 0 {
            let _ = self.tx.send(event());
        }
    }
}

impl<Ctx: Context> Default for TxEvent<Ctx> {
    fn default() -> Self {
        Self::new()
    }
}

#[derive_where(Clone, Debug)]
pub enum Event<Ctx: Context> {
    StartedHeight(Ctx::Height, bool),
    StartedRound(Ctx::Height, Round, Ctx::Address, Role),
    Published(SignedConsensusMsg<Ctx>),
    Received(SignedConsensusMsg<Ctx>),
    ProposedValue(LocallyProposedValue<Ctx>),
    ReceivedProposedValue(ProposedValue<Ctx>, ValueOrigin),
    Decided {
        commit_certificate: CommitCertificate<Ctx>,
    },
    Finalized {
        commit_certificate: CommitCertificate<Ctx>,
        evidence: MisbehaviorEvidence<Ctx>,
    },
    RepublishVote(SignedVote<Ctx>),
    RebroadcastRoundCertificate(RoundCertificate<Ctx>),
    SkipRoundCertificate(RoundCertificate<Ctx>),
    PolkaCertificate(PolkaCertificate<Ctx>),
    WalReplayBegin(Ctx::Height, usize),
    WalReplayEntry(WalEntry<Ctx>),
    WalReplayDone(Ctx::Height),
    WalReplayError(Arc<ConsensusError<Ctx>>),
    WalResetError(Arc<eyre::Report>),
    WalCorrupted(Arc<io::Error>),
}

impl<Ctx: Context> fmt::Display for Event<Ctx> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Event::StartedHeight(height, restart) => {
                write!(f, "StartedHeight(height: {height}, restart: {restart})")
            }
            Event::StartedRound(height, round, proposer, role) => {
                write!(f, "StartedRound(height: {height}, round: {round}, proposer: {proposer}, role: {role:?})")
            }
            Event::Published(msg) => write!(f, "Published(msg: {msg:?})"),
            Event::Received(msg) => write!(f, "Received(msg: {msg:?})"),
            Event::ProposedValue(value) => write!(f, "ProposedValue(value: {value:?})"),
            Event::ReceivedProposedValue(value, origin) => {
                write!(
                    f,
                    "ReceivedProposedValue(value: {value:?}, origin: {origin:?})"
                )
            }
            Event::Decided { commit_certificate } => {
                write!(
                    f,
                    "Decided(value: {}, signatures: {})",
                    commit_certificate.value_id,
                    commit_certificate.commit_signatures.len()
                )
            }
            Event::Finalized {
                commit_certificate,
                evidence,
            } => {
                if evidence.is_empty() {
                    write!(
                        f,
                        "Finalized(value: {}, signatures: {})",
                        commit_certificate.value_id,
                        commit_certificate.commit_signatures.len()
                    )
                } else {
                    write!(
                        f,
                        "Finalized(value: {}, signatures: {}, evidence: {:?})",
                        commit_certificate.value_id,
                        commit_certificate.commit_signatures.len(),
                        evidence
                    )
                }
            }
            Event::RepublishVote(vote) => write!(f, "RepublishVote(vote: {vote:?})"),
            Event::RebroadcastRoundCertificate(certificate) => write!(
                f,
                "RebroadcastRoundCertificate(certificate: {certificate:?})"
            ),
            Event::WalReplayBegin(height, count) => {
                write!(f, "WalReplayBegin(height: {height}, count: {count})")
            }
            Event::WalReplayEntry(entry) => write!(f, "WalReplayEntry(entry: {entry:?})"),
            Event::WalReplayDone(height) => write!(f, "WalReplayDone(height: {height})"),
            Event::WalReplayError(error) => write!(f, "WalReplayError({error})"),
            Event::WalResetError(error) => write!(f, "WalResetError({error})"),
            Event::WalCorrupted(error) => write!(f, "WalCorrupted(error: {error:?})"),

            Event::PolkaCertificate(certificate) => {
                write!(f, "PolkaCertificate: {certificate:?})")
            }
            Event::SkipRoundCertificate(certificate) => {
                write!(f, "SkipRoundCertificate: {certificate:?})")
            }
        }
    }
}