mirror of
https://github.com/serai-dex/serai.git
synced 2025-12-10 21:19:24 +00:00
3c664ff05f
coordinator/tributary was tributary-chain. This crate has been renamed tributary-sdk and moved to coordinator/tributary-sdk. coordinator/src/tributary was our instantion of a Tributary, the Transaction type and scan task. This has been moved to coordinator/tributary. The main reason for this was due to coordinator/main.rs becoming untidy. There is now a collection of clean, independent APIs present in the codebase. coordinator/main.rs is to compose them. Sometimes, these compositions are a bit silly (reading from a channel just to forward the message to a distinct channel). That's more than fine as the code is still readable and the value from the cleanliness of the APIs composed far exceeds the nits from having these odd compositions. This breaks down a bit as we now define a global database, and have some APIs interact with multiple other APIs. coordinator/src/tributary was a self-contained, clean API. The recently added task present in coordinator/tributary/mod.rs, which bound it to the rest of the Coordinator, wasn't. Now, coordinator/src is solely the API compositions, and all self-contained APIs are their own crates.
1093 lines
36 KiB
Rust
1093 lines
36 KiB
Rust
use core::fmt::Debug;
|
|
|
|
use std::{
|
|
sync::Arc,
|
|
time::{SystemTime, Instant, Duration},
|
|
collections::{VecDeque, HashMap},
|
|
};
|
|
|
|
use parity_scale_codec::{Encode, Decode, IoReader};
|
|
|
|
use futures_channel::mpsc;
|
|
use futures_util::{
|
|
FutureExt, StreamExt, SinkExt,
|
|
future::{self, Fuse},
|
|
};
|
|
use patchable_async_sleep::sleep;
|
|
|
|
use serai_db::{Get, DbTxn, Db};
|
|
|
|
pub mod time;
|
|
use time::{sys_time, CanonicalInstant};
|
|
|
|
pub mod round;
|
|
use round::RoundData;
|
|
|
|
mod block;
|
|
use block::BlockData;
|
|
|
|
pub(crate) mod message_log;
|
|
|
|
/// Traits and types of the external network being integrated with to provide consensus over.
|
|
pub mod ext;
|
|
use ext::*;
|
|
|
|
const MESSAGE_TAPE_KEY: &[u8] = b"tendermint-machine-message_tape";
|
|
fn message_tape_key(genesis: [u8; 32]) -> Vec<u8> {
|
|
[MESSAGE_TAPE_KEY, &genesis].concat()
|
|
}
|
|
|
|
pub fn commit_msg(end_time: u64, id: &[u8]) -> Vec<u8> {
|
|
[&end_time.to_le_bytes(), id].concat()
|
|
}
|
|
|
|
#[derive(Clone, Copy, PartialEq, Eq, Hash, Debug, Encode, Decode)]
|
|
pub enum Step {
|
|
Propose,
|
|
Prevote,
|
|
Precommit,
|
|
}
|
|
|
|
#[derive(Clone, Eq, Debug, Encode, Decode)]
|
|
pub enum Data<B: Block, S: Signature> {
|
|
Proposal(Option<RoundNumber>, B),
|
|
Prevote(Option<B::Id>),
|
|
Precommit(Option<(B::Id, S)>),
|
|
}
|
|
|
|
impl<B: Block, S: Signature> PartialEq for Data<B, S> {
|
|
fn eq(&self, other: &Data<B, S>) -> bool {
|
|
match (self, other) {
|
|
(Data::Proposal(valid_round, block), Data::Proposal(valid_round2, block2)) => {
|
|
(valid_round == valid_round2) && (block == block2)
|
|
}
|
|
(Data::Prevote(id), Data::Prevote(id2)) => id == id2,
|
|
(Data::Precommit(None), Data::Precommit(None)) => true,
|
|
(Data::Precommit(Some((id, _))), Data::Precommit(Some((id2, _)))) => id == id2,
|
|
_ => false,
|
|
}
|
|
}
|
|
}
|
|
|
|
impl<B: Block, S: Signature> core::hash::Hash for Data<B, S> {
|
|
fn hash<H: core::hash::Hasher>(&self, state: &mut H) {
|
|
match self {
|
|
Data::Proposal(valid_round, block) => (0, valid_round, block.id().as_ref()).hash(state),
|
|
Data::Prevote(id) => (1, id.as_ref().map(AsRef::<[u8]>::as_ref)).hash(state),
|
|
Data::Precommit(None) => (2, 0).hash(state),
|
|
Data::Precommit(Some((id, _))) => (2, 1, id.as_ref()).hash(state),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl<B: Block, S: Signature> Data<B, S> {
|
|
pub fn step(&self) -> Step {
|
|
match self {
|
|
Data::Proposal(..) => Step::Propose,
|
|
Data::Prevote(..) => Step::Prevote,
|
|
Data::Precommit(..) => Step::Precommit,
|
|
}
|
|
}
|
|
}
|
|
|
|
#[derive(Clone, PartialEq, Eq, Debug, Encode, Decode)]
|
|
pub struct Message<V: ValidatorId, B: Block, S: Signature> {
|
|
pub sender: V,
|
|
pub block: BlockNumber,
|
|
pub round: RoundNumber,
|
|
|
|
pub data: Data<B, S>,
|
|
}
|
|
|
|
/// A signed Tendermint consensus message to be broadcast to the other validators.
|
|
#[derive(Clone, PartialEq, Eq, Debug, Encode, Decode)]
|
|
pub struct SignedMessage<V: ValidatorId, B: Block, S: Signature> {
|
|
pub msg: Message<V, B, S>,
|
|
pub sig: S,
|
|
}
|
|
|
|
impl<V: ValidatorId, B: Block, S: Signature> SignedMessage<V, B, S> {
|
|
/// Number of the block this message is attempting to add to the chain.
|
|
pub fn block(&self) -> BlockNumber {
|
|
self.msg.block
|
|
}
|
|
|
|
#[must_use]
|
|
pub fn verify_signature<Scheme: SignatureScheme<ValidatorId = V, Signature = S>>(
|
|
&self,
|
|
signer: &Scheme,
|
|
) -> bool {
|
|
signer.verify(self.msg.sender, &self.msg.encode(), &self.sig)
|
|
}
|
|
}
|
|
|
|
#[derive(Clone, Copy, PartialEq, Eq, Debug, Encode, Decode)]
|
|
pub enum SlashReason {
|
|
FailToPropose,
|
|
InvalidBlock,
|
|
InvalidProposer,
|
|
}
|
|
|
|
#[derive(Clone, PartialEq, Eq, Debug, Encode, Decode)]
|
|
pub enum Evidence {
|
|
ConflictingMessages(Vec<u8>, Vec<u8>),
|
|
InvalidPrecommit(Vec<u8>),
|
|
InvalidValidRound(Vec<u8>),
|
|
}
|
|
|
|
#[derive(Clone, PartialEq, Eq, Debug)]
|
|
pub enum TendermintError {
|
|
Malicious,
|
|
Temporal,
|
|
AlreadyHandled,
|
|
InvalidEvidence,
|
|
}
|
|
|
|
// Type aliases to abstract over generic hell
|
|
pub type DataFor<N> =
|
|
Data<<N as Network>::Block, <<N as Network>::SignatureScheme as SignatureScheme>::Signature>;
|
|
pub(crate) type MessageFor<N> = Message<
|
|
<N as Network>::ValidatorId,
|
|
<N as Network>::Block,
|
|
<<N as Network>::SignatureScheme as SignatureScheme>::Signature,
|
|
>;
|
|
/// Type alias to the SignedMessage type for a given Network
|
|
pub type SignedMessageFor<N> = SignedMessage<
|
|
<N as Network>::ValidatorId,
|
|
<N as Network>::Block,
|
|
<<N as Network>::SignatureScheme as SignatureScheme>::Signature,
|
|
>;
|
|
|
|
pub fn decode_signed_message<N: Network>(mut data: &[u8]) -> Option<SignedMessageFor<N>> {
|
|
SignedMessageFor::<N>::decode(&mut data).ok()
|
|
}
|
|
|
|
fn decode_and_verify_signed_message<N: Network>(
|
|
data: &[u8],
|
|
schema: &N::SignatureScheme,
|
|
) -> Result<SignedMessageFor<N>, TendermintError> {
|
|
let msg = decode_signed_message::<N>(data).ok_or(TendermintError::InvalidEvidence)?;
|
|
|
|
// verify that evidence messages are signed correctly
|
|
if !msg.verify_signature(schema) {
|
|
Err(TendermintError::InvalidEvidence)?;
|
|
}
|
|
|
|
Ok(msg)
|
|
}
|
|
|
|
pub fn verify_tendermint_evidence<N: Network>(
|
|
evidence: &Evidence,
|
|
schema: &N::SignatureScheme,
|
|
commit: impl Fn(u64) -> Option<Commit<N::SignatureScheme>>,
|
|
) -> Result<(), TendermintError> {
|
|
match evidence {
|
|
Evidence::ConflictingMessages(first, second) => {
|
|
let first = decode_and_verify_signed_message::<N>(first, schema)?.msg;
|
|
let second = decode_and_verify_signed_message::<N>(second, schema)?.msg;
|
|
|
|
// Make sure they're distinct messages, from the same sender, within the same block
|
|
if (first == second) || (first.sender != second.sender) || (first.block != second.block) {
|
|
Err(TendermintError::InvalidEvidence)?;
|
|
}
|
|
|
|
// Distinct messages within the same step
|
|
if !((first.round == second.round) && (first.data.step() == second.data.step())) {
|
|
Err(TendermintError::InvalidEvidence)?;
|
|
}
|
|
}
|
|
Evidence::InvalidPrecommit(msg) => {
|
|
let msg = decode_and_verify_signed_message::<N>(msg, schema)?.msg;
|
|
|
|
let Data::Precommit(Some((id, sig))) = &msg.data else {
|
|
Err(TendermintError::InvalidEvidence)?
|
|
};
|
|
// TODO: We need to be passed in the genesis time to handle this edge case
|
|
if msg.block.0 == 0 {
|
|
Err(TendermintError::InvalidEvidence)?
|
|
// todo!("invalid precommit signature on first block")
|
|
}
|
|
|
|
// get the last commit
|
|
let prior_commit = match commit(msg.block.0 - 1) {
|
|
Some(c) => c,
|
|
// If we have yet to sync the block in question, we will return InvalidEvidence based
|
|
// on our own temporal ambiguity
|
|
// This will also cause an InvalidEvidence for anything using a non-existent block,
|
|
// yet that's valid behavior
|
|
// TODO: Double check the ramifications of this
|
|
_ => Err(TendermintError::InvalidEvidence)?,
|
|
};
|
|
|
|
// calculate the end time till the msg round
|
|
let mut last_end_time = CanonicalInstant::new(prior_commit.end_time);
|
|
for r in 0 ..= msg.round.0 {
|
|
last_end_time = RoundData::<N>::new(RoundNumber(r), last_end_time).end_time();
|
|
}
|
|
|
|
// verify that the commit was actually invalid
|
|
if schema.verify(msg.sender, &commit_msg(last_end_time.canonical(), id.as_ref()), sig) {
|
|
Err(TendermintError::InvalidEvidence)?
|
|
}
|
|
}
|
|
Evidence::InvalidValidRound(msg) => {
|
|
let msg = decode_and_verify_signed_message::<N>(msg, schema)?.msg;
|
|
|
|
let Data::Proposal(Some(vr), _) = &msg.data else { Err(TendermintError::InvalidEvidence)? };
|
|
if vr.0 < msg.round.0 {
|
|
Err(TendermintError::InvalidEvidence)?
|
|
}
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
#[derive(Clone, PartialEq, Eq, Debug)]
|
|
pub enum SlashEvent {
|
|
Id(SlashReason, u64, u32),
|
|
WithEvidence(Evidence),
|
|
}
|
|
|
|
// Struct for if various upon handlers have been triggered to ensure they don't trigger multiple
|
|
// times.
|
|
#[derive(Clone, PartialEq, Eq, Debug)]
|
|
struct Upons {
|
|
upon_prevotes: bool,
|
|
upon_successful_current_round_prevotes: bool,
|
|
upon_negative_current_round_prevotes: bool,
|
|
upon_precommits: bool,
|
|
}
|
|
|
|
/// A machine executing the Tendermint protocol.
|
|
pub struct TendermintMachine<N: Network> {
|
|
db: N::Db,
|
|
genesis: [u8; 32],
|
|
|
|
network: N,
|
|
signer: <N::SignatureScheme as SignatureScheme>::Signer,
|
|
validators: N::SignatureScheme,
|
|
weights: Arc<N::Weights>,
|
|
|
|
queue: VecDeque<MessageFor<N>>,
|
|
msg_recv: mpsc::UnboundedReceiver<SignedMessageFor<N>>,
|
|
synced_block_recv: mpsc::UnboundedReceiver<SyncedBlock<N>>,
|
|
synced_block_result_send: mpsc::UnboundedSender<bool>,
|
|
|
|
block: BlockData<N>,
|
|
// TODO: Move this into the Block struct
|
|
round_proposals: HashMap<RoundNumber, (Option<RoundNumber>, N::Block)>,
|
|
// TODO: Move this into the Round struct
|
|
upons: Upons,
|
|
}
|
|
|
|
pub struct SyncedBlock<N: Network> {
|
|
pub number: BlockNumber,
|
|
pub block: <N as Network>::Block,
|
|
pub commit: Commit<<N as Network>::SignatureScheme>,
|
|
}
|
|
|
|
pub type SyncedBlockSender<N> = mpsc::UnboundedSender<SyncedBlock<N>>;
|
|
pub type SyncedBlockResultReceiver = mpsc::UnboundedReceiver<bool>;
|
|
|
|
pub type MessageSender<N> = mpsc::UnboundedSender<SignedMessageFor<N>>;
|
|
|
|
/// A Tendermint machine and its channel to receive messages from the gossip layer over.
|
|
pub struct TendermintHandle<N: Network> {
|
|
/// Channel to trigger the machine to move to the next block.
|
|
/// Takes in the the previous block's commit, along with the new proposal.
|
|
pub synced_block: SyncedBlockSender<N>,
|
|
/// A channel to communicate the result of a synced_block message.
|
|
pub synced_block_result: SyncedBlockResultReceiver,
|
|
/// Channel to send messages received from the P2P layer.
|
|
pub messages: MessageSender<N>,
|
|
/// Tendermint machine to be run on an asynchronous task.
|
|
pub machine: TendermintMachine<N>,
|
|
}
|
|
|
|
impl<N: Network + 'static> TendermintMachine<N> {
|
|
// Broadcast the given piece of data
|
|
// Tendermint messages always specify their block/round, yet Tendermint only ever broadcasts for
|
|
// the current block/round. Accordingly, instead of manually fetching those at every call-site,
|
|
// this function can simply pass the data to the block which can contextualize it
|
|
fn broadcast(&mut self, data: DataFor<N>) {
|
|
if let Some(msg) = self.block.message(data) {
|
|
// Push it on to the queue. This is done so we only handle one message at a time, and so we
|
|
// can handle our own message before broadcasting it. That way, we fail before before
|
|
// becoming malicious
|
|
self.queue.push_back(msg);
|
|
}
|
|
}
|
|
|
|
// Start a new round. Returns true if we were the proposer
|
|
fn round(&mut self, round: RoundNumber, time: Option<CanonicalInstant>) -> bool {
|
|
// Clear upons
|
|
self.upons = Upons {
|
|
upon_prevotes: false,
|
|
upon_successful_current_round_prevotes: false,
|
|
upon_negative_current_round_prevotes: false,
|
|
upon_precommits: false,
|
|
};
|
|
|
|
let proposer = self.weights.proposer(self.block.number, round);
|
|
let res = if let Some(data) = self.block.new_round(round, proposer, time) {
|
|
self.broadcast(data);
|
|
true
|
|
} else {
|
|
false
|
|
};
|
|
log::debug!(
|
|
target: "tendermint",
|
|
"proposer for block {}, round {round:?} was {} (me: {res})",
|
|
self.block.number.0,
|
|
hex::encode(proposer.encode()),
|
|
);
|
|
res
|
|
}
|
|
|
|
// 53-54
|
|
async fn reset(&mut self, end_round: RoundNumber, proposal: Option<N::Block>) {
|
|
// Ensure we have the end time data for the last round
|
|
self.block.populate_end_time(end_round);
|
|
|
|
// Sleep until this round ends
|
|
let round_end = self.block.end_time[&end_round];
|
|
let time_until_round_end = round_end.instant().saturating_duration_since(Instant::now());
|
|
if time_until_round_end == Duration::ZERO {
|
|
log::trace!(
|
|
target: "tendermint",
|
|
"resetting when prior round ended {}ms ago",
|
|
Instant::now().saturating_duration_since(round_end.instant()).as_millis(),
|
|
);
|
|
}
|
|
log::trace!(
|
|
target: "tendermint",
|
|
"sleeping until round ends in {}ms",
|
|
time_until_round_end.as_millis(),
|
|
);
|
|
sleep(time_until_round_end).await;
|
|
|
|
// Clear the message tape
|
|
{
|
|
let mut txn = self.db.txn();
|
|
txn.del(message_tape_key(self.genesis));
|
|
txn.commit();
|
|
}
|
|
|
|
// Clear our outbound message queue
|
|
self.queue = VecDeque::new();
|
|
|
|
// Create the new block
|
|
self.block = BlockData::new(
|
|
self.db.clone(),
|
|
self.genesis,
|
|
self.weights.clone(),
|
|
BlockNumber(self.block.number.0 + 1),
|
|
self.signer.validator_id().await,
|
|
proposal,
|
|
);
|
|
|
|
// Reset the round proposals
|
|
self.round_proposals = HashMap::new();
|
|
|
|
// Start the first round
|
|
self.round(RoundNumber(0), Some(round_end));
|
|
}
|
|
|
|
async fn reset_by_commit(
|
|
&mut self,
|
|
commit: Commit<N::SignatureScheme>,
|
|
proposal: Option<N::Block>,
|
|
) {
|
|
let mut round = self.block.round().number;
|
|
// If this commit is for a round we don't have, jump up to it
|
|
while self.block.end_time[&round].canonical() < commit.end_time {
|
|
round.0 += 1;
|
|
self.block.populate_end_time(round);
|
|
}
|
|
// If this commit is for a prior round, find it
|
|
while self.block.end_time[&round].canonical() > commit.end_time {
|
|
if round.0 == 0 {
|
|
panic!("commit isn't for this machine's next block");
|
|
}
|
|
round.0 -= 1;
|
|
}
|
|
debug_assert_eq!(self.block.end_time[&round].canonical(), commit.end_time);
|
|
|
|
self.reset(round, proposal).await;
|
|
}
|
|
|
|
async fn slash(&mut self, validator: N::ValidatorId, slash_event: SlashEvent) {
|
|
// TODO: If the new slash event has evidence, emit to prevent a low-importance slash from
|
|
// cancelling emission of high-importance slashes
|
|
if !self.block.slashes.contains(&validator) {
|
|
log::info!(target: "tendermint", "Slashing validator {}", hex::encode(validator.encode()));
|
|
self.block.slashes.insert(validator);
|
|
self.network.slash(validator, slash_event).await;
|
|
}
|
|
}
|
|
|
|
fn proposal_for_round(&self, round: RoundNumber) -> Option<(Option<RoundNumber>, &N::Block)> {
|
|
self.round_proposals.get(&round).map(|(round, block)| (*round, block))
|
|
}
|
|
|
|
// L22-27
|
|
fn upon_proposal_without_valid_round(&mut self) {
|
|
if self.block.round().step != Step::Propose {
|
|
return;
|
|
}
|
|
|
|
// If we have the proposal message...
|
|
let Some((None, block)) = self.proposal_for_round(self.block.round().number) else {
|
|
return;
|
|
};
|
|
|
|
// There either needs to not be a locked value or it must be equivalent
|
|
#[allow(clippy::map_unwrap_or)]
|
|
if self
|
|
.block
|
|
.locked
|
|
.as_ref()
|
|
.map(|(_round, locked_block)| block.id() == *locked_block)
|
|
.unwrap_or(true)
|
|
{
|
|
self.broadcast(Data::Prevote(Some(block.id())));
|
|
} else {
|
|
self.broadcast(Data::Prevote(None));
|
|
}
|
|
}
|
|
|
|
// L28-33
|
|
fn upon_proposal_with_valid_round(&mut self) {
|
|
if self.block.round().step != Step::Propose {
|
|
return;
|
|
}
|
|
|
|
// If we have the proposal message...
|
|
let Some((Some(proposal_valid_round), block)) =
|
|
self.proposal_for_round(self.block.round().number)
|
|
else {
|
|
return;
|
|
};
|
|
|
|
// Check we have the necessary prevotes
|
|
if !self.block.log.has_consensus(proposal_valid_round, &Data::Prevote(Some(block.id()))) {
|
|
return;
|
|
}
|
|
|
|
// We don't check valid round < current round as the `message` function does
|
|
|
|
// If locked is None, lockedRoundp is -1 and less than valid round
|
|
#[allow(clippy::map_unwrap_or)]
|
|
let locked_clause_1 = self
|
|
.block
|
|
.locked
|
|
.as_ref()
|
|
.map(|(locked_round, _block)| locked_round.0 <= proposal_valid_round.0)
|
|
.unwrap_or(true);
|
|
// The second clause is if the locked values are equivalent. If no value is locked, they aren't
|
|
#[allow(clippy::map_unwrap_or)]
|
|
let locked_clause_2 = self
|
|
.block
|
|
.locked
|
|
.as_ref()
|
|
.map(|(_round, locked_block)| block.id() == *locked_block)
|
|
.unwrap_or(false);
|
|
|
|
if locked_clause_1 || locked_clause_2 {
|
|
self.broadcast(Data::Prevote(Some(block.id())));
|
|
} else {
|
|
self.broadcast(Data::Prevote(None));
|
|
}
|
|
}
|
|
|
|
// L34-35
|
|
fn upon_prevotes(&mut self) {
|
|
if self.upons.upon_prevotes || (self.block.round().step != Step::Prevote) {
|
|
return;
|
|
}
|
|
|
|
if self.block.log.has_participation(self.block.round().number, Step::Prevote) {
|
|
self.block.round_mut().set_timeout(Step::Prevote);
|
|
self.upons.upon_prevotes = true;
|
|
}
|
|
}
|
|
|
|
// L36-43
|
|
async fn upon_successful_current_round_prevotes(&mut self) {
|
|
// Returning if `self.step == Step::Propose` is equivalent to guarding `step >= prevote`
|
|
if self.upons.upon_successful_current_round_prevotes ||
|
|
(self.block.round().step == Step::Propose)
|
|
{
|
|
return;
|
|
}
|
|
|
|
// If we have the proposal message...
|
|
let Some((_, block)) = self.proposal_for_round(self.block.round().number) else {
|
|
return;
|
|
};
|
|
|
|
// Check we have the necessary prevotes
|
|
if !self.block.log.has_consensus(self.block.round().number, &Data::Prevote(Some(block.id()))) {
|
|
return;
|
|
}
|
|
|
|
let block = block.clone();
|
|
self.upons.upon_successful_current_round_prevotes = true;
|
|
|
|
if self.block.round().step == Step::Prevote {
|
|
self.block.locked = Some((self.block.round().number, block.id()));
|
|
let signature = self
|
|
.signer
|
|
.sign(&commit_msg(
|
|
self.block.end_time[&self.block.round().number].canonical(),
|
|
block.id().as_ref(),
|
|
))
|
|
.await;
|
|
self.broadcast(Data::Precommit(Some((block.id(), signature))));
|
|
}
|
|
self.block.valid = Some((self.block.round().number, block));
|
|
}
|
|
|
|
// L44-46
|
|
fn upon_negative_current_round_prevotes(&mut self) {
|
|
if self.upons.upon_negative_current_round_prevotes || (self.block.round().step != Step::Prevote)
|
|
{
|
|
return;
|
|
}
|
|
|
|
if self.block.log.has_consensus(self.block.round().number, &Data::Prevote(None)) {
|
|
self.broadcast(Data::Precommit(None));
|
|
}
|
|
|
|
self.upons.upon_negative_current_round_prevotes = true;
|
|
}
|
|
|
|
// L47-48
|
|
fn upon_precommits(&mut self) {
|
|
if self.upons.upon_precommits {
|
|
return;
|
|
}
|
|
|
|
if self.block.log.has_participation(self.block.round().number, Step::Precommit) {
|
|
self.block.round_mut().set_timeout(Step::Precommit);
|
|
self.upons.upon_precommits = true;
|
|
}
|
|
}
|
|
|
|
// L22-48
|
|
async fn all_current_round_upons(&mut self) {
|
|
self.upon_proposal_without_valid_round();
|
|
self.upon_proposal_with_valid_round();
|
|
self.upon_prevotes();
|
|
self.upon_successful_current_round_prevotes().await;
|
|
self.upon_negative_current_round_prevotes();
|
|
self.upon_precommits();
|
|
}
|
|
|
|
// L49-54
|
|
async fn upon_successful_precommits(&mut self, round: RoundNumber) -> bool {
|
|
// If we have the proposal message...
|
|
let Some((_, block)) = self.proposal_for_round(round) else { return false };
|
|
|
|
// Check we have the necessary precommits
|
|
// The precommit we check we have consensus upon uses a junk signature since message equality
|
|
// disregards the signature
|
|
if !self
|
|
.block
|
|
.log
|
|
.has_consensus(round, &Data::Precommit(Some((block.id(), self.signer.sign(&[]).await))))
|
|
{
|
|
return false;
|
|
}
|
|
|
|
// Get all participants in this commit
|
|
let mut validators = vec![];
|
|
let mut sigs = vec![];
|
|
// Get all precommits for this round
|
|
for (validator, msgs) in &self.block.log.log[&round] {
|
|
if let Some(signed) = msgs.get(&Step::Precommit) {
|
|
if let Data::Precommit(Some((id, sig))) = &signed.msg.data {
|
|
// If this precommit was for this block, include it
|
|
if *id == block.id() {
|
|
validators.push(*validator);
|
|
sigs.push(sig.clone());
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Form the commit itself
|
|
let commit_msg = commit_msg(self.block.end_time[&round].canonical(), block.id().as_ref());
|
|
let commit = Commit {
|
|
end_time: self.block.end_time[&round].canonical(),
|
|
validators: validators.clone(),
|
|
signature: self.network.signature_scheme().aggregate(&validators, &commit_msg, &sigs),
|
|
};
|
|
debug_assert!(self.network.verify_commit(block.id(), &commit));
|
|
|
|
// Add the block and reset the machine
|
|
log::info!(
|
|
target: "tendermint",
|
|
"TendermintMachine produced block {}",
|
|
hex::encode(block.id().as_ref()),
|
|
);
|
|
let id = block.id();
|
|
let proposal = self.network.add_block(block.clone(), commit).await;
|
|
log::trace!(
|
|
target: "tendermint",
|
|
"added block {} (produced by machine)",
|
|
hex::encode(id.as_ref()),
|
|
);
|
|
self.reset(round, proposal).await;
|
|
|
|
true
|
|
}
|
|
|
|
// L49-54
|
|
async fn all_any_round_upons(&mut self, round: RoundNumber) -> bool {
|
|
self.upon_successful_precommits(round).await
|
|
}
|
|
|
|
// Returns Ok(true) if this was a Precommit which had either no signature or its signature
|
|
// validated
|
|
// Returns Ok(false) if it wasn't a Precommit or the signature wasn't validated yet
|
|
// Returns Err if the signature was invalid
|
|
async fn verify_precommit_signature(
|
|
&mut self,
|
|
signed: &SignedMessageFor<N>,
|
|
) -> Result<bool, TendermintError> {
|
|
let msg = &signed.msg;
|
|
if let Data::Precommit(precommit) = &msg.data {
|
|
let Some((id, sig)) = precommit else { return Ok(true) };
|
|
// Also verify the end_time of the commit
|
|
// Only perform this verification if we already have the end_time
|
|
// Else, there's a DoS where we receive a precommit for some round infinitely in the future
|
|
// which forces us to calculate every end time
|
|
if let Some(end_time) = self.block.end_time.get(&msg.round) {
|
|
if !self.validators.verify(msg.sender, &commit_msg(end_time.canonical(), id.as_ref()), sig)
|
|
{
|
|
log::warn!(target: "tendermint", "validator produced an invalid commit signature");
|
|
self
|
|
.slash(
|
|
msg.sender,
|
|
SlashEvent::WithEvidence(Evidence::InvalidPrecommit(signed.encode())),
|
|
)
|
|
.await;
|
|
Err(TendermintError::Malicious)?;
|
|
}
|
|
return Ok(true);
|
|
}
|
|
}
|
|
Ok(false)
|
|
}
|
|
|
|
async fn message(&mut self, signed: &SignedMessageFor<N>) -> Result<(), TendermintError> {
|
|
let msg = &signed.msg;
|
|
if msg.block != self.block.number {
|
|
Err(TendermintError::Temporal)?;
|
|
}
|
|
|
|
// If this is a precommit, verify its signature
|
|
self.verify_precommit_signature(signed).await?;
|
|
|
|
// Only let the proposer propose
|
|
if matches!(msg.data, Data::Proposal(..)) &&
|
|
(msg.sender != self.weights.proposer(msg.block, msg.round))
|
|
{
|
|
log::warn!(target: "tendermint", "validator who wasn't the proposer proposed");
|
|
// TODO: This should have evidence
|
|
self
|
|
.slash(msg.sender, SlashEvent::Id(SlashReason::InvalidProposer, msg.block.0, msg.round.0))
|
|
.await;
|
|
Err(TendermintError::Malicious)?;
|
|
};
|
|
|
|
// If this is a proposal, verify the block
|
|
// If the block is invalid, drop the message, letting the timeout cover it
|
|
// This prevents needing to check if valid inside every `upon` block
|
|
if let Data::Proposal(_, block) = &msg.data {
|
|
match self.network.validate(block).await {
|
|
Ok(()) => {}
|
|
Err(BlockError::Temporal) => {
|
|
if self.block.round().step == Step::Propose {
|
|
self.broadcast(Data::Prevote(None));
|
|
}
|
|
Err(TendermintError::Temporal)?;
|
|
}
|
|
Err(BlockError::Fatal) => {
|
|
log::warn!(target: "tendermint", "validator proposed a fatally invalid block");
|
|
if self.block.round().step == Step::Propose {
|
|
self.broadcast(Data::Prevote(None));
|
|
}
|
|
self
|
|
.slash(
|
|
msg.sender,
|
|
SlashEvent::Id(SlashReason::InvalidBlock, self.block.number.0, msg.round.0),
|
|
)
|
|
.await;
|
|
Err(TendermintError::Malicious)?;
|
|
}
|
|
};
|
|
}
|
|
|
|
// If this is a proposal, verify the valid round isn't fundamentally invalid
|
|
if let Data::Proposal(Some(valid_round), _) = msg.data {
|
|
if valid_round.0 >= msg.round.0 {
|
|
log::warn!(
|
|
target: "tendermint",
|
|
"proposed proposed with a syntactically invalid valid round",
|
|
);
|
|
if self.block.round().step == Step::Propose {
|
|
self.broadcast(Data::Prevote(None));
|
|
}
|
|
self
|
|
.slash(msg.sender, SlashEvent::WithEvidence(Evidence::InvalidValidRound(msg.encode())))
|
|
.await;
|
|
Err(TendermintError::Malicious)?;
|
|
}
|
|
}
|
|
|
|
// Add it to the log, returning if it was already handled
|
|
match self.block.log.log(signed.clone()) {
|
|
Ok(true) => {}
|
|
Ok(false) => Err(TendermintError::AlreadyHandled)?,
|
|
Err(evidence) => {
|
|
self.slash(msg.sender, SlashEvent::WithEvidence(evidence)).await;
|
|
Err(TendermintError::Malicious)?;
|
|
}
|
|
}
|
|
log::debug!(
|
|
target: "tendermint",
|
|
"received new tendermint message (block: {}, round: {}, step: {:?})",
|
|
msg.block.0,
|
|
msg.round.0,
|
|
msg.data.step(),
|
|
);
|
|
|
|
// If this is a proposal, insert it
|
|
if let Data::Proposal(vr, block) = &msg.data {
|
|
self.round_proposals.insert(msg.round, (*vr, block.clone()));
|
|
}
|
|
|
|
// L55-56
|
|
// Jump ahead if we should
|
|
if (msg.round.0 > self.block.round().number.0) &&
|
|
(self.block.log.round_participation(msg.round) >= self.weights.fault_threshold())
|
|
{
|
|
log::debug!(
|
|
target: "tendermint",
|
|
"jumping from round {} to round {}",
|
|
self.block.round().number.0,
|
|
msg.round.0,
|
|
);
|
|
|
|
// Jump to the new round.
|
|
let old_round = self.block.round().number;
|
|
self.round(msg.round, None);
|
|
|
|
// If any jumped over/to round already has precommit messages, verify their signatures
|
|
for jumped in (old_round.0 + 1) ..= msg.round.0 {
|
|
let jumped = RoundNumber(jumped);
|
|
let round_msgs = self.block.log.log.get(&jumped).cloned().unwrap_or_default();
|
|
for (validator, msgs) in &round_msgs {
|
|
if let Some(existing) = msgs.get(&Step::Precommit) {
|
|
if let Ok(res) = self.verify_precommit_signature(existing).await {
|
|
// Ensure this actually verified the signature instead of believing it shouldn't yet
|
|
assert!(res);
|
|
} else {
|
|
// Remove the message so it isn't counted towards forming a commit/included in one
|
|
// This won't remove the fact they precommitted for this block hash in the MessageLog
|
|
// TODO: Don't even log these in the first place until we jump, preventing needing
|
|
// to do this in the first place
|
|
self
|
|
.block
|
|
.log
|
|
.log
|
|
.get_mut(&jumped)
|
|
.unwrap()
|
|
.get_mut(validator)
|
|
.unwrap()
|
|
.remove(&Step::Precommit)
|
|
.unwrap();
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Now that we've jumped, and:
|
|
// 1) If this is a message for an old round, verified the precommit signatures
|
|
// 2) If this is a message for what was the current round, verified the precommit signatures
|
|
// 3) If this is a message for what was a future round, verified the precommit signatures if it
|
|
// has 34+% participation
|
|
// Run all `upons` run for any round, which may produce a Commit if it has 67+% participation
|
|
// (returning true if it does, letting us return now)
|
|
// It's necessary to verify the precommit signatures before Commit production is allowed, hence
|
|
// this specific flow
|
|
if self.all_any_round_upons(msg.round).await {
|
|
return Ok(());
|
|
}
|
|
|
|
// If this is a historic round, or a future round without sufficient participation, return
|
|
if msg.round.0 != self.block.round().number.0 {
|
|
return Ok(());
|
|
}
|
|
// msg.round is now guaranteed to be equal to self.block.round().number
|
|
debug_assert_eq!(msg.round, self.block.round().number);
|
|
|
|
// Run all `upons` run for the current round
|
|
self.all_current_round_upons().await;
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Create a new Tendermint machine, from the specified point, with the specified block as the
|
|
/// one to propose next. This will return a channel to send messages from the gossip layer and
|
|
/// the machine itself. The machine should have `run` called from an asynchronous task.
|
|
#[allow(clippy::new_ret_no_self)]
|
|
pub async fn new(
|
|
db: N::Db,
|
|
network: N,
|
|
genesis: [u8; 32],
|
|
last_block: BlockNumber,
|
|
last_time: u64,
|
|
proposal: N::Block,
|
|
) -> TendermintHandle<N> {
|
|
let (msg_send, msg_recv) = mpsc::unbounded();
|
|
let (synced_block_send, synced_block_recv) = mpsc::unbounded();
|
|
let (synced_block_result_send, synced_block_result_recv) = mpsc::unbounded();
|
|
TendermintHandle {
|
|
synced_block: synced_block_send,
|
|
synced_block_result: synced_block_result_recv,
|
|
messages: msg_send,
|
|
machine: {
|
|
let now = SystemTime::now();
|
|
let sys_time = sys_time(last_time);
|
|
let mut negative = false;
|
|
let time_until = sys_time.duration_since(now).unwrap_or_else(|_| {
|
|
negative = true;
|
|
now.duration_since(sys_time).unwrap_or(Duration::ZERO)
|
|
});
|
|
log::info!(
|
|
target: "tendermint",
|
|
"new TendermintMachine building off block {} is scheduled to start in {}{}s",
|
|
last_block.0,
|
|
if negative { "-" } else { "" },
|
|
time_until.as_secs(),
|
|
);
|
|
|
|
// If the last block hasn't ended yet, sleep until it has
|
|
if !negative {
|
|
sleep(time_until).await;
|
|
}
|
|
|
|
let signer = network.signer();
|
|
let validators = network.signature_scheme();
|
|
let weights = Arc::new(network.weights());
|
|
let validator_id = signer.validator_id().await;
|
|
// L01-10
|
|
let mut machine = TendermintMachine {
|
|
db: db.clone(),
|
|
genesis,
|
|
|
|
network,
|
|
signer,
|
|
validators,
|
|
weights: weights.clone(),
|
|
|
|
queue: VecDeque::new(),
|
|
msg_recv,
|
|
synced_block_recv,
|
|
synced_block_result_send,
|
|
|
|
block: BlockData::new(
|
|
db,
|
|
genesis,
|
|
weights,
|
|
BlockNumber(last_block.0 + 1),
|
|
validator_id,
|
|
Some(proposal),
|
|
),
|
|
|
|
round_proposals: HashMap::new(),
|
|
|
|
upons: Upons {
|
|
upon_prevotes: false,
|
|
upon_successful_current_round_prevotes: false,
|
|
upon_negative_current_round_prevotes: false,
|
|
upon_precommits: false,
|
|
},
|
|
};
|
|
|
|
// The end time of the last block is the start time for this one
|
|
// The Commit explicitly contains the end time, so loading the last commit will provide
|
|
// this. The only exception is for the genesis block, which doesn't have a commit
|
|
// Using the genesis time in place will cause this block to be created immediately
|
|
// after it, without the standard amount of separation (so their times will be
|
|
// equivalent or minimally offset)
|
|
// For callers wishing to avoid this, they should pass (0, GENESIS + N::block_time())
|
|
machine.round(RoundNumber(0), Some(CanonicalInstant::new(last_time)));
|
|
machine
|
|
},
|
|
}
|
|
}
|
|
|
|
pub async fn run(mut self) {
|
|
log::debug!(target: "tendermint", "running TendermintMachine");
|
|
|
|
let mut rebroadcast_future = Box::pin(sleep(Duration::from_secs(60))).fuse();
|
|
loop {
|
|
// Also create a future for if the queue has a message
|
|
// Does not pop_front as if another message has higher priority, its future will be handled
|
|
// instead in this loop, and the popped value would be dropped with the next iteration
|
|
let mut queue_future =
|
|
if self.queue.is_empty() { Fuse::terminated() } else { future::ready(()).fuse() };
|
|
|
|
if let Some((our_message, msg, mut sig)) = futures_util::select_biased! {
|
|
// Handle a new block occurring externally (from an external sync loop)
|
|
// Has the highest priority as it makes all other futures here irrelevant
|
|
msg = self.synced_block_recv.next() => {
|
|
if let Some(SyncedBlock { number, block, commit }) = msg {
|
|
// Commit is for a block we've already moved past
|
|
if number != self.block.number {
|
|
self.synced_block_result_send.send(false).await.unwrap();
|
|
continue;
|
|
}
|
|
|
|
// Commit is invalid
|
|
if !self.network.verify_commit(block.id(), &commit) {
|
|
self.synced_block_result_send.send(false).await.unwrap();
|
|
continue;
|
|
}
|
|
|
|
log::debug!(
|
|
target: "tendermint",
|
|
"TendermintMachine received a block from the external sync loop",
|
|
);
|
|
let proposal = self.network.add_block(block, commit.clone()).await;
|
|
self.reset_by_commit(commit, proposal).await;
|
|
self.synced_block_result_send.send(true).await.unwrap();
|
|
None
|
|
} else {
|
|
break;
|
|
}
|
|
},
|
|
|
|
// Handle our messages
|
|
() = queue_future => {
|
|
Some((true, self.queue.pop_front().unwrap(), None))
|
|
},
|
|
|
|
// L57-67
|
|
// Handle any timeouts
|
|
step = self.block.round().timeout_future().fuse() => {
|
|
// Remove the timeout so it doesn't persist, always being the selected future due to bias
|
|
// While this does enable the timeout to be entered again, the timeout setting code will
|
|
// never attempt to add a timeout after its timeout has expired
|
|
// (due to it setting an `upon` boolean)
|
|
self.block.round_mut().timeouts.remove(&step);
|
|
|
|
match step {
|
|
Step::Propose => {
|
|
// Only run if it's still the step in question
|
|
if self.block.round().step == step {
|
|
// Slash the validator for not proposing when they should've
|
|
log::debug!(target: "tendermint", "validator didn't propose when they should have");
|
|
// this slash will be voted on.
|
|
self.slash(
|
|
self.weights.proposer(self.block.number, self.block.round().number),
|
|
SlashEvent::Id(
|
|
SlashReason::FailToPropose,
|
|
self.block.number.0,
|
|
self.block.round().number.0
|
|
),
|
|
).await;
|
|
self.broadcast(Data::Prevote(None));
|
|
}
|
|
},
|
|
Step::Prevote => {
|
|
// Only run if it's still the step in question
|
|
if self.block.round().step == step {
|
|
self.broadcast(Data::Precommit(None))
|
|
}
|
|
},
|
|
Step::Precommit => {
|
|
self.round(RoundNumber(self.block.round().number.0 + 1), None);
|
|
}
|
|
};
|
|
|
|
// Execute the upons now that the state has changed
|
|
self.all_any_round_upons(self.block.round().number).await;
|
|
self.all_current_round_upons().await;
|
|
|
|
None
|
|
},
|
|
|
|
// If it's been more than 60s, rebroadcast our own messages
|
|
() = rebroadcast_future => {
|
|
log::trace!("rebroadcast future hit within tendermint machine");
|
|
let key = message_tape_key(self.genesis);
|
|
let messages = self.db.get(key).unwrap_or(vec![]);
|
|
let mut messages = messages.as_slice();
|
|
|
|
while !messages.is_empty() {
|
|
self.network.broadcast(
|
|
SignedMessageFor::<N>::decode(&mut IoReader(&mut messages))
|
|
.expect("saved invalid message to DB")
|
|
).await;
|
|
}
|
|
|
|
// Reset the rebroadcast future
|
|
rebroadcast_future = Box::pin(sleep(core::time::Duration::from_secs(60))).fuse();
|
|
|
|
None
|
|
},
|
|
|
|
// Handle any received messages
|
|
msg = self.msg_recv.next() => {
|
|
if let Some(msg) = msg {
|
|
if !msg.verify_signature(&self.validators) {
|
|
continue;
|
|
}
|
|
Some((false, msg.msg, Some(msg.sig)))
|
|
} else {
|
|
break;
|
|
}
|
|
}
|
|
} {
|
|
if our_message {
|
|
assert!(sig.is_none());
|
|
sig = Some(self.signer.sign(&msg.encode()).await);
|
|
}
|
|
let sig = sig.unwrap();
|
|
|
|
let signed_msg = SignedMessage { msg: msg.clone(), sig: sig.clone() };
|
|
let res = self.message(&signed_msg).await;
|
|
// If this is our message, and we hit an invariant, we could be slashed.
|
|
// We only broadcast our message after running it ourselves, to ensure it doesn't error, to
|
|
// ensure we don't get slashed on invariants.
|
|
if res.is_err() && our_message {
|
|
panic!("honest node (ourselves) had invalid behavior");
|
|
}
|
|
|
|
// Save this message to a linear tape of all our messages for this block, if ours
|
|
// TODO: Since we do this after we mark this message as sent to prevent equivocations, a
|
|
// precisely time reboot could cause this message marked as sent yet not added to the tape
|
|
if our_message {
|
|
let message_tape_key = message_tape_key(self.genesis);
|
|
let mut txn = self.db.txn();
|
|
let mut message_tape = txn.get(&message_tape_key).unwrap_or(vec![]);
|
|
message_tape.extend(signed_msg.encode());
|
|
txn.put(&message_tape_key, message_tape);
|
|
txn.commit();
|
|
}
|
|
|
|
// Re-broadcast this since it's an original consensus message worth handling
|
|
if res.is_ok() {
|
|
self.network.broadcast(signed_msg).await;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|