use std::{collections::HashMap, time::Duration};
use crate::{
error::ConsensusError,
protos::consensus::v1::{Proposal, Vote},
scope_config::{NetworkType, ScopeConfig},
signing::ConsensusSignatureScheme,
types::SessionTransition,
utils::{
calculate_consensus_result, calculate_max_rounds, validate_proposal,
validate_proposal_timestamp, validate_vote, validate_vote_chain,
},
};
#[derive(Debug, Clone)]
pub struct ConsensusConfig {
consensus_threshold: f64,
consensus_timeout: Duration,
max_rounds: u32,
use_gossipsub_rounds: bool,
liveness_criteria: bool,
}
impl From<NetworkType> for ConsensusConfig {
fn from(network_type: NetworkType) -> Self {
ConsensusConfig::from(ScopeConfig::from(network_type))
}
}
impl From<ScopeConfig> for ConsensusConfig {
fn from(config: ScopeConfig) -> Self {
let (max_rounds, use_gossipsub_rounds) = match config.network_type {
NetworkType::Gossipsub => (config.max_rounds_override.unwrap_or(2), true),
NetworkType::P2P => (config.max_rounds_override.unwrap_or(0), false),
};
ConsensusConfig::new(
config.default_consensus_threshold,
config.default_timeout,
max_rounds,
use_gossipsub_rounds,
config.default_liveness_criteria_yes,
)
}
}
impl ConsensusConfig {
pub fn p2p() -> Self {
ConsensusConfig::from(NetworkType::P2P)
}
pub fn gossipsub() -> Self {
ConsensusConfig::from(NetworkType::Gossipsub)
}
pub fn with_timeout(mut self, consensus_timeout: Duration) -> Result<Self, ConsensusError> {
crate::utils::validate_timeout(consensus_timeout)?;
self.consensus_timeout = consensus_timeout;
Ok(self)
}
pub fn with_threshold(mut self, consensus_threshold: f64) -> Result<Self, ConsensusError> {
crate::utils::validate_threshold(consensus_threshold)?;
self.consensus_threshold = consensus_threshold;
Ok(self)
}
pub fn with_liveness_criteria(mut self, liveness_criteria: bool) -> Self {
self.liveness_criteria = liveness_criteria;
self
}
pub(crate) fn new(
consensus_threshold: f64,
consensus_timeout: Duration,
max_rounds: u32,
use_gossipsub_rounds: bool,
liveness_criteria: bool,
) -> Self {
Self {
consensus_threshold,
consensus_timeout,
max_rounds,
use_gossipsub_rounds,
liveness_criteria,
}
}
fn max_round_limit(&self, expected_voters_count: u32) -> u32 {
if self.use_gossipsub_rounds {
self.max_rounds
} else if self.max_rounds == 0 {
calculate_max_rounds(expected_voters_count, self.consensus_threshold)
} else {
self.max_rounds
}
}
pub fn consensus_timeout(&self) -> Duration {
self.consensus_timeout
}
pub fn consensus_threshold(&self) -> f64 {
self.consensus_threshold
}
pub fn liveness_criteria(&self) -> bool {
self.liveness_criteria
}
pub fn max_rounds(&self) -> u32 {
self.max_rounds
}
pub fn use_gossipsub_rounds(&self) -> bool {
self.use_gossipsub_rounds
}
}
#[derive(Debug, Clone)]
pub enum ConsensusState {
Active,
ConsensusReached(bool),
Failed,
}
#[derive(Debug, Clone)]
pub struct ConsensusSession {
pub proposal: Proposal,
pub state: ConsensusState,
pub votes: HashMap<Vec<u8>, Vote>, pub created_at: u64,
pub config: ConsensusConfig,
}
impl ConsensusSession {
fn new(proposal: Proposal, config: ConsensusConfig, now: u64) -> Self {
Self {
proposal,
state: ConsensusState::Active,
votes: HashMap::new(),
created_at: now,
config,
}
}
pub fn from_proposal<Signer: ConsensusSignatureScheme>(
proposal: Proposal,
config: ConsensusConfig,
now: u64,
) -> Result<(Self, SessionTransition), ConsensusError> {
validate_proposal::<Signer>(&proposal, now)?;
let existing_votes = proposal.votes.clone();
let mut clean_proposal = proposal.clone();
clean_proposal.votes.clear();
clean_proposal.round = 1;
let mut session = Self::new(clean_proposal, config, now);
let transition = session.initialize_with_votes::<Signer>(
existing_votes,
proposal.expiration_timestamp,
proposal.timestamp,
now,
)?;
Ok((session, transition))
}
pub(crate) fn add_vote(
&mut self,
vote: Vote,
now: u64,
) -> Result<SessionTransition, ConsensusError> {
match self.state {
ConsensusState::Active => {
validate_proposal_timestamp(self.proposal.expiration_timestamp, now)?;
self.check_round_limit(1)?;
if self.votes.contains_key(&vote.vote_owner) {
return Err(ConsensusError::DuplicateVote);
}
self.votes.insert(vote.vote_owner.clone(), vote.clone());
self.proposal.votes.push(vote.clone());
self.update_round(1);
Ok(self.check_consensus())
}
ConsensusState::ConsensusReached(res) => Ok(SessionTransition::ConsensusReached(res)),
_ => Err(ConsensusError::SessionNotActive),
}
}
pub(crate) fn initialize_with_votes<Signer: ConsensusSignatureScheme>(
&mut self,
votes: Vec<Vote>,
expiration_timestamp: u64,
creation_time: u64,
now: u64,
) -> Result<SessionTransition, ConsensusError> {
if !matches!(self.state, ConsensusState::Active) {
return Err(ConsensusError::SessionNotActive);
}
validate_proposal_timestamp(expiration_timestamp, now)?;
if votes.is_empty() {
return Ok(SessionTransition::StillActive);
}
let mut seen_owners = std::collections::HashSet::new();
for vote in &votes {
if !seen_owners.insert(&vote.vote_owner) {
return Err(ConsensusError::DuplicateVote);
}
}
if votes.len() > self.proposal.expected_voters_count as usize {
self.state = ConsensusState::Failed;
return Err(ConsensusError::MaxRoundsExceeded);
}
validate_vote_chain(&votes)?;
for vote in &votes {
validate_vote::<Signer>(vote, expiration_timestamp, creation_time, now)?;
}
self.check_round_limit(votes.len())?;
self.update_round(votes.len());
for vote in votes {
self.votes.insert(vote.vote_owner.clone(), vote.clone());
self.proposal.votes.push(vote);
}
Ok(self.check_consensus())
}
fn check_round_limit(&mut self, vote_count: usize) -> Result<(), ConsensusError> {
if vote_count > self.proposal.expected_voters_count as usize {
self.state = ConsensusState::Failed;
return Err(ConsensusError::MaxRoundsExceeded);
}
let projected_value = if self.config.use_gossipsub_rounds {
if self.proposal.round == 2 || (self.proposal.round == 1 && vote_count > 0) {
2
} else {
self.proposal.round }
} else {
let current_votes = self.proposal.round.saturating_sub(1);
current_votes.saturating_add(vote_count as u32)
};
if projected_value
> self
.config
.max_round_limit(self.proposal.expected_voters_count)
{
self.state = ConsensusState::Failed;
return Err(ConsensusError::MaxRoundsExceeded);
}
Ok(())
}
fn update_round(&mut self, vote_count: usize) {
if self.config.use_gossipsub_rounds {
if self.proposal.round == 1 && vote_count > 0 {
self.proposal.round = 2;
}
} else {
self.proposal.round = self.proposal.round.saturating_add(vote_count as u32);
}
}
fn check_consensus(&mut self) -> SessionTransition {
let expected_voters = self.proposal.expected_voters_count;
let threshold = self.config.consensus_threshold;
let liveness = self.proposal.liveness_criteria_yes;
match calculate_consensus_result(&self.votes, expected_voters, threshold, liveness, false) {
Some(result) => {
self.state = ConsensusState::ConsensusReached(result);
SessionTransition::ConsensusReached(result)
}
None => {
self.state = ConsensusState::Active;
SessionTransition::StillActive
}
}
}
pub fn is_active(&self) -> bool {
matches!(self.state, ConsensusState::Active)
}
pub fn get_consensus_result(&self) -> Result<bool, ConsensusError> {
if let ConsensusState::ConsensusReached(result) = self.state {
Ok(result)
} else {
Err(ConsensusError::ConsensusNotReached)
}
}
}
#[cfg(test)]
mod tests {
use std::time::Duration;
use alloy::signers::local::PrivateKeySigner;
use crate::{
error::ConsensusError,
session::{ConsensusConfig, ConsensusSession, ConsensusState},
signing::EthereumConsensusSigner,
test_utils::now_ts,
types::CreateProposalRequest,
utils::build_vote,
};
fn wrap(signer: PrivateKeySigner) -> EthereumConsensusSigner {
EthereumConsensusSigner::new(signer)
}
#[test]
fn enforce_max_rounds_gossipsub() {
let signer1 = PrivateKeySigner::random();
let signer2 = PrivateKeySigner::random();
let signer3 = PrivateKeySigner::random();
let signer4 = PrivateKeySigner::random();
let request = CreateProposalRequest::new(
"Test".into(),
"".into(),
signer1.address().as_slice().to_vec(),
4, 60,
false,
)
.unwrap();
let proposal = request.into_proposal(now_ts()).unwrap();
let config = ConsensusConfig::gossipsub();
let mut session = ConsensusSession::new(proposal, config, now_ts());
let vote1 = build_vote(&session.proposal, true, &wrap(signer1), now_ts()).unwrap();
session.add_vote(vote1, now_ts()).unwrap();
assert_eq!(session.proposal.round, 2);
let vote2 = build_vote(&session.proposal, false, &wrap(signer2), now_ts()).unwrap();
session.add_vote(vote2, now_ts()).unwrap();
assert_eq!(session.proposal.round, 2);
let vote3 = build_vote(&session.proposal, true, &wrap(signer3), now_ts()).unwrap();
session.add_vote(vote3, now_ts()).unwrap();
assert_eq!(session.proposal.round, 2);
let vote4 = build_vote(&session.proposal, true, &wrap(signer4), now_ts()).unwrap();
session.add_vote(vote4, now_ts()).unwrap();
assert_eq!(session.proposal.round, 2);
assert_eq!(session.votes.len(), 4);
}
#[test]
fn enforce_max_rounds_p2p() {
let signer1 = PrivateKeySigner::random();
let signer2 = PrivateKeySigner::random();
let signer3 = PrivateKeySigner::random();
let signer4 = PrivateKeySigner::random();
let signer5 = PrivateKeySigner::random();
let request = CreateProposalRequest::new(
"Test".into(),
"".into(),
signer1.address().as_slice().to_vec(),
5,
60,
false,
)
.unwrap();
let proposal = request.into_proposal(now_ts()).unwrap();
let config = ConsensusConfig::p2p();
let mut session = ConsensusSession::new(proposal, config, now_ts());
let vote1 = build_vote(&session.proposal, true, &wrap(signer1), now_ts()).unwrap();
session.add_vote(vote1, now_ts()).unwrap();
assert_eq!(session.proposal.round, 2);
assert_eq!(session.votes.len(), 1);
let vote2 = build_vote(&session.proposal, false, &wrap(signer2), now_ts()).unwrap();
session.add_vote(vote2, now_ts()).unwrap();
assert_eq!(session.proposal.round, 3);
assert_eq!(session.votes.len(), 2);
let vote3 = build_vote(&session.proposal, true, &wrap(signer3), now_ts()).unwrap();
session.add_vote(vote3, now_ts()).unwrap();
assert_eq!(session.proposal.round, 4);
assert_eq!(session.votes.len(), 3);
let vote4 = build_vote(&session.proposal, true, &wrap(signer4), now_ts()).unwrap();
session.add_vote(vote4, now_ts()).unwrap();
assert_eq!(session.proposal.round, 5);
assert_eq!(session.votes.len(), 4);
let vote5 = build_vote(&session.proposal, true, &wrap(signer5), now_ts()).unwrap();
let err = session.add_vote(vote5, now_ts()).unwrap_err();
assert!(matches!(err, ConsensusError::MaxRoundsExceeded));
}
#[test]
fn consensus_config_builder_and_getters_cover_edges() {
let cfg = ConsensusConfig::gossipsub()
.with_threshold(0.75)
.unwrap()
.with_timeout(Duration::from_secs(42))
.unwrap()
.with_liveness_criteria(false);
assert_eq!(cfg.consensus_threshold(), 0.75);
assert_eq!(cfg.consensus_timeout(), Duration::from_secs(42));
assert!(!cfg.liveness_criteria());
let err = ConsensusConfig::gossipsub()
.with_threshold(1.1)
.unwrap_err();
assert!(matches!(err, ConsensusError::InvalidConsensusThreshold));
let err = ConsensusConfig::gossipsub()
.with_timeout(Duration::from_secs(0))
.unwrap_err();
assert!(matches!(err, ConsensusError::InvalidTimeout));
let explicit = ConsensusConfig::new(2.0 / 3.0, Duration::from_secs(60), 7, false, true);
assert_eq!(explicit.max_round_limit(100), 7);
}
#[test]
fn add_vote_rejects_non_active_and_reports_reached_when_finalized() {
let signer = PrivateKeySigner::random();
let request = CreateProposalRequest::new(
"Test".into(),
"".into(),
signer.address().as_slice().to_vec(),
3,
60,
true,
)
.unwrap();
let proposal = request.into_proposal(now_ts()).unwrap();
let mut failed_session =
ConsensusSession::new(proposal.clone(), ConsensusConfig::gossipsub(), now_ts());
failed_session.state = ConsensusState::Failed;
let vote = build_vote(
&failed_session.proposal,
true,
&wrap(signer.clone()),
now_ts(),
)
.unwrap();
let err = failed_session.add_vote(vote, now_ts()).unwrap_err();
assert!(matches!(err, ConsensusError::SessionNotActive));
let mut finalized_session =
ConsensusSession::new(proposal, ConsensusConfig::gossipsub(), now_ts());
finalized_session.state = ConsensusState::ConsensusReached(true);
let vote = build_vote(&finalized_session.proposal, true, &wrap(signer), now_ts()).unwrap();
let transition = finalized_session.add_vote(vote, now_ts()).unwrap();
assert!(matches!(
transition,
crate::types::SessionTransition::ConsensusReached(true)
));
}
#[test]
fn initialize_with_votes_non_active_duplicate_and_zero_votes_paths() {
let signer = PrivateKeySigner::random();
let request = CreateProposalRequest::new(
"Test".into(),
"".into(),
signer.address().as_slice().to_vec(),
4,
60,
true,
)
.unwrap();
let proposal = request.into_proposal(now_ts()).unwrap();
let mut inactive =
ConsensusSession::new(proposal.clone(), ConsensusConfig::gossipsub(), now_ts());
inactive.state = ConsensusState::Failed;
let err = inactive
.initialize_with_votes::<EthereumConsensusSigner>(
vec![],
proposal.expiration_timestamp,
proposal.timestamp,
now_ts(),
)
.unwrap_err();
assert!(matches!(err, ConsensusError::SessionNotActive));
let mut dup_session =
ConsensusSession::new(proposal.clone(), ConsensusConfig::gossipsub(), now_ts());
let vote1 =
build_vote(&dup_session.proposal, true, &wrap(signer.clone()), now_ts()).unwrap();
let vote2 = build_vote(&dup_session.proposal, false, &wrap(signer), now_ts()).unwrap();
let err = dup_session
.initialize_with_votes::<EthereumConsensusSigner>(
vec![vote1, vote2],
proposal.expiration_timestamp,
proposal.timestamp,
now_ts(),
)
.unwrap_err();
assert!(matches!(err, ConsensusError::DuplicateVote));
let mut zero_votes =
ConsensusSession::new(proposal, ConsensusConfig::gossipsub(), now_ts());
zero_votes.check_round_limit(0).unwrap();
}
#[test]
fn p2p_round_limit_should_reject_effectively_huge_vote_count() {
if usize::BITS <= 32 {
return;
}
let signer = PrivateKeySigner::random();
let request = CreateProposalRequest::new(
"TruncationTest".into(),
vec![],
signer.address().as_slice().to_vec(),
1,
60,
true,
)
.unwrap();
let proposal = request.into_proposal(now_ts()).unwrap();
let mut session = ConsensusSession::new(proposal, ConsensusConfig::p2p(), now_ts());
let wrapped_vote_count = (u32::MAX as usize) + 1;
let result = session.check_round_limit(wrapped_vote_count);
assert!(
result.is_err(),
"effectively huge vote_count should not pass round-limit checks"
);
}
#[test]
fn p2p_update_round_should_advance_for_max_u32_vote_count() {
let signer = PrivateKeySigner::random();
let request = CreateProposalRequest::new(
"RoundUpdateMax".into(),
vec![],
signer.address().as_slice().to_vec(),
u32::MAX,
60,
true,
)
.unwrap();
let proposal = request.into_proposal(now_ts()).unwrap();
let mut session = ConsensusSession::new(proposal, ConsensusConfig::p2p(), now_ts());
let starting_round = session.proposal.round;
session.update_round(u32::MAX as usize);
assert!(
session.proposal.round > starting_round,
"round should advance when max u32 vote_count is applied"
);
assert_eq!(session.proposal.round, u32::MAX);
}
}