blockchain/Consensus.cpp¶
Consensus proposal/vote/certificate helpers. More...
Namespaces¶
| Name |
|---|
| sgns |
Functions¶
| Name | |
|---|---|
| base::Logger | ConsensusManagerLogger() |
Detailed Description¶
Consensus proposal/vote/certificate helpers.
Date: 2025-10-16 Henrique A. Klein ([email protected])
Functions Documentation¶
function ConsensusManagerLogger¶
Source code¶
#include "blockchain/Consensus.hpp"
#include <algorithm>
#include <chrono>
#include <regex>
#include <set>
#include <system_error>
#include <boost/format.hpp>
#include <gsl/span>
#include "base/hexutil.hpp"
#include "base/sgns_version.hpp"
#include "crypto/hasher.hpp"
#include "account/GeniusAccount.hpp"
#include "blockchain/ConsensusAuth.hpp"
#include "storage/database_error.hpp"
namespace sgns
{
namespace
{
bool AddCandidateIfAbsent( std::vector<ConsensusManager::Proposal> &candidates,
const ConsensusManager::Proposal &candidate )
{
if ( std::any_of( candidates.begin(),
candidates.end(),
[&candidate]( const ConsensusManager::Proposal &known )
{ return known.proposal_id() == candidate.proposal_id(); } ) )
{
return false;
}
candidates.push_back( candidate );
return true;
}
uint64_t ActiveVoterWeight( const ValidatorRegistry::Registry ®istry, const std::string &voter_id )
{
const auto *validator = ValidatorRegistry::FindValidator( registry, voter_id );
if ( !validator || validator->status() != ValidatorRegistry::Status::ACTIVE )
{
return 0;
}
return validator->weight();
}
bool IsPublicChainMintChainId( std::string_view chain_id )
{
if ( chain_id.empty() )
{
return false;
}
if ( chain_id == "public" )
{
return true;
}
return std::all_of( chain_id.begin(),
chain_id.end(),
[]( unsigned char c ) { return c >= '0' && c <= '9'; } );
}
uint64_t SlotEvidenceReputation( const std::vector<ConsensusVote> &votes,
const ValidatorRegistry::Registry ®istry )
{
std::unordered_set<std::string> seen;
uint64_t reputation = 0;
for ( const auto &vote : votes )
{
if ( !vote.approve() || !seen.insert( vote.voter_id() ).second )
{
continue;
}
const bool carries_slot_hash = !vote.slot_0_hash().empty() || !vote.slot_1_hash().empty() ||
!vote.slot_2_hash().empty();
if ( !carries_slot_hash )
{
continue;
}
const auto *validator = ValidatorRegistry::FindValidator( registry, vote.voter_id() );
if ( validator && validator->status() == ValidatorRegistry::Status::ACTIVE )
{
reputation += validator->weight();
}
}
return reputation;
}
std::string SerializedCertificateHash( std::string_view serialized )
{
const auto hash = crypto::sha2_256( serialized.data(), serialized.size() );
return base::hex_lower( gsl::span<const uint8_t>( hash.data(), hash.size() ) );
}
std::optional<std::string> VerifiedMintV2TransactionHash( const ConsensusManager::Certificate &certificate )
{
if ( !certificate.has_proposal() )
{
return std::nullopt;
}
const auto nonce = ConsensusManager::DecodeNonceSubject( certificate.proposal().subject() );
if ( nonce.has_error() || nonce.value().tx_hash().empty() ||
nonce.value().transaction().transaction_case() != EmbeddedTransaction::kMintV2 )
{
return std::nullopt;
}
return nonce.value().tx_hash();
}
// Registry batch subjects must carry the canonical certificate slot of every
// member; hash-only batch subjects fail closed (CERT-02 slot authority).
// Callers log the rejection reason.
bool RegistryBatchSubjectMembersValid( const RegistryBatchSubject &payload )
{
const auto slot_count = payload.member_certificate_slots_size();
if ( slot_count == 0 || static_cast<uint32_t>( slot_count ) != payload.certificate_count() )
{
return false;
}
std::vector<std::string> slots;
slots.reserve( static_cast<size_t>( slot_count ) );
for ( const auto &slot : payload.member_certificate_slots() )
{
if ( slot.empty() )
{
return false;
}
slots.push_back( slot );
}
auto root = ConsensusManager::ComputeBatchRoot( slots );
if ( root.has_error() || root.value() != std::string( payload.batch_root() ) )
{
return false;
}
return true;
}
} // namespace
base::Logger ConsensusManagerLogger()
{
// Always call base::createLogger to get the current logger
// This will return existing logger or create new one as needed
return base::createLogger( "ConsensusManager" );
}
namespace
{
std::string ConsensusManagerLoggerName( const std::string &address )
{
return "ConsensusManager:" + address.substr( 0, 8 );
}
base::Logger MakeConsensusManagerLogger( const std::string &address )
{
return ConsensusManagerLogger()->clone( ConsensusManagerLoggerName( address ) );
}
}
std::shared_ptr<ConsensusManager> ConsensusManager::New( std::shared_ptr<ValidatorRegistry> registry,
std::shared_ptr<crdt::GlobalDB> db,
std::shared_ptr<ipfs_pubsub::GossipPubSub> pubsub,
Signer signer,
std::string address,
std::string consensus_topic,
NodeType node_type )
{
if ( !registry )
{
ConsensusManagerLogger()->error( "{}: Failed to create ConsensusManager: registry is null", __func__ );
return nullptr;
}
if ( !db )
{
ConsensusManagerLogger()->error( "{}: Failed to create ConsensusManager: db is null", __func__ );
return nullptr;
}
if ( !pubsub )
{
ConsensusManagerLogger()->error( "{}: Failed to create ConsensusManager: pubsub is null", __func__ );
return nullptr;
}
if ( !signer )
{
ConsensusManagerLogger()->error( "{}: Failed to create ConsensusManager: signer is null", __func__ );
return nullptr;
}
if ( address.empty() )
{
ConsensusManagerLogger()->error( "{}: Failed to create ConsensusManager: address is empty", __func__ );
return nullptr;
}
auto instance = std::shared_ptr<ConsensusManager>( new ConsensusManager( std::move( registry ),
std::move( db ),
std::move( pubsub ),
std::move( signer ),
std::move( address ),
std::move( consensus_topic ),
node_type ) );
instance->certificate_work_journal_ = instance->db_->GetWorkJournal();
if ( !instance->certificate_work_journal_ )
{
ConsensusManagerLogger()->error( "{}: Failed to create ConsensusManager: crdt work journal is empty",
__func__ );
return nullptr;
}
instance->consensus_subs_future_ = std::move( instance->pubsub_->Subscribe(
instance->consensus_messages_topic_,
[weakptr( std::weak_ptr<ConsensusManager>( instance ) )](
boost::optional<const ipfs_pubsub::GossipPubSub::Message &> message )
{
if ( auto self = weakptr.lock() )
{
ConsensusManagerLogger()->trace( "{}: Received Consensus Message on topic {}",
__func__,
self->consensus_messages_topic_ );
self->OnConsensusMessage( message );
}
} ) );
ConsensusManagerLogger()->debug( "{}: Subscribed to Consensus topic {}",
__func__,
instance->consensus_messages_topic_ );
ConsensusManagerLogger()->info( "{}: role={} self-voting={}",
__func__,
NodeTypeToString( node_type ),
instance->participates_in_consensus_ ? "enabled" : "disabled" );
instance->RecoverActiveVotes();
instance->StartRoundTimer();
if ( !instance->RegisterCertificateFilter() )
{
ConsensusManagerLogger()->error( "{}: Failed to register certificate filter", __func__ );
}
instance->RecoverPendingCertificateWork();
return instance;
}
ConsensusManager::ConsensusManager( std::shared_ptr<ValidatorRegistry> registry,
std::shared_ptr<crdt::GlobalDB> db,
std::shared_ptr<ipfs_pubsub::GossipPubSub> pubsub,
Signer signer,
std::string address,
std::string consensus_topic,
NodeType node_type ) :
registry_( std::move( registry ) ), //
db_( std::move( db ) ), //
pubsub_( std::move( pubsub ) ), //
signer_( std::move( signer ) ), //
account_address_( std::move( address ) ), //
logger_( MakeConsensusManagerLogger( account_address_ ) ),
// ::sgns:: qualified: the member accessor of the same name would otherwise shadow the
// free predicate in class scope.
participates_in_consensus_( ::sgns::ParticipatesInConsensus( node_type ) ),
consensus_messages_topic_( std::string( CONSENSUS_CHANNEL_PREFIX ) + sgns::version::GetNetAndVersionAppendix() +
consensus_topic ),
consensus_datastore_topic_( consensus_messages_topic_ + "#datastore" )
{
}
ConsensusManager::~ConsensusManager()
{
Close();
logger_->debug( "{}: Finished shutting down ConsensusManager", __func__ );
}
void ConsensusManager::Close()
{
bool expected = false;
if ( !close_started_.compare_exchange_strong( expected, true ) )
{
return;
}
// Account switches reuse GlobalDB. Remove this manager's registrations
// before a replacement ConsensusManager registers the same pattern.
// The one-shot guard also prevents a delayed destructor from removing
// registrations that belong to the replacement manager.
const std::string pattern = std::string( CERT_KEY_PATTERN );
if ( db_ )
{
if ( certificate_callback_registered_ )
{
db_->UnregisterNewElementCallback( pattern );
certificate_callback_registered_ = false;
}
if ( certificate_filter_registered_ )
{
db_->UnregisterElementFilter( pattern );
certificate_filter_registered_ = false;
}
}
stop_timer_.store( true );
timer_cv_.notify_all();
fault_test_cv_.notify_all();
std::thread timer;
{
std::lock_guard lock( close_mutex_ );
if ( !round_timer_.joinable() )
{
return;
}
if ( round_timer_.get_id() == std::this_thread::get_id() )
{
// A timer callback can release the final owner. A thread cannot
// join itself, so detach only in this self-teardown path.
round_timer_.detach();
return;
}
timer = std::move( round_timer_ );
}
timer.join();
}
void ConsensusManager::StartRoundTimer()
{
std::lock_guard close_lock( close_mutex_ );
if ( round_timer_.joinable() || stop_timer_.load() )
{
return;
}
std::weak_ptr<ConsensusManager> weak_self = shared_from_this();
round_timer_ = std::thread(
[weak_self]()
{
constexpr auto min_interval = std::chrono::milliseconds( 500 );
while ( true )
{
auto self = weak_self.lock();
if ( !self )
{
return;
}
std::unique_lock<std::mutex> lock( self->timer_mutex_ );
auto interval = std::chrono::milliseconds(
self->round_duration_ms_.load( std::memory_order_relaxed ) ) / 2;
if ( interval.count() <= 0 )
{
interval = DEFAULT_ROUND_DURATION / 2;
}
if ( interval < min_interval )
{
interval = min_interval;
}
if ( self->certificates_pending_.load() )
{
// Work is pending: run on cadence, only interrupt for shutdown.
self->timer_cv_.wait_for( lock, interval, [self]() { return self->stop_timer_.load(); } );
}
else
{
// No pending work: wait up to interval, but wake immediately when new work appears.
self->timer_cv_.wait_for(
lock,
interval,
[self]() { return self->stop_timer_.load() || self->certificates_pending_.load(); } );
}
if ( self->stop_timer_.load() )
{
return;
}
lock.unlock();
std::function<void()> timer_work_hook;
{
std::lock_guard timer_lock( self->timer_mutex_ );
timer_work_hook = self->timer_work_hook_for_test_;
}
if ( timer_work_hook )
{
timer_work_hook();
}
if ( self->certificates_pending_.load() )
{
self->ProcessCertificates();
self->UpdateCertificatesPending();
}
self->ExpirePendingProposals();
self->ProcessDuePendingRetries();
self->ProcessDueVoteWork();
// Keep replaying unfinished certificate work while the node is running.
self->RecoverPendingCertificateWork();
}
} );
}
outcome::result<void> ConsensusManager::Publish( const ConsensusMessage &message )
{
std::vector<uint8_t> serialized_proto( message.ByteSizeLong() );
if ( !message.SerializeToArray( serialized_proto.data(), serialized_proto.size() ) )
{
logger_->error( "{}: Failed to serialize consensus message", __func__ );
return outcome::failure( std::errc::invalid_argument );
}
logger_->debug( "{}: Sending consensus packet to {}", __func__, consensus_messages_topic_ );
auto publish_result = pubsub_->Publish( consensus_messages_topic_, serialized_proto );
if ( publish_result.has_error() )
{
return outcome::failure( publish_result.error() );
}
logger_->debug( "{}: Consensus packet published (bytes={})",
__func__,
serialized_proto.size() );
if ( message.has_certificate() )
{
std::lock_guard lock( fault_test_mutex_ );
++fault_test_counters_.certificate_notification_publications;
}
return outcome::success();
}
bool ConsensusManager::RegisterSubjectHandler( std::string_view subject_type, SubjectHandler handler )
{
if ( !handler )
{
logger_->error( "{}: ignored empty handler subject_type={}", __func__, subject_type );
return false;
}
auto type_hash = ComputeSubjectTypeHash( subject_type );
if ( type_hash.has_error() )
{
logger_->error( "{}: ignored invalid handler subject_type={}", __func__, subject_type );
return false;
}
logger_->debug( "{}: Registering subject handler subject_type={}", __func__, subject_type );
std::unique_lock lock( subject_handlers_mutex_ );
subject_handlers_[type_hash.value()] = std::move( handler );
return true;
}
void ConsensusManager::UnregisterSubjectHandler( std::string_view subject_type )
{
logger_->debug( "{}: Removing Subject handler with subject_type={}", __func__, subject_type );
auto type_hash = ComputeSubjectTypeHash( subject_type );
if ( type_hash.has_error() )
{
return;
}
std::unique_lock lock( subject_handlers_mutex_ );
subject_handlers_.erase( type_hash.value() );
}
bool ConsensusManager::RegisterCertificateHandler( std::string_view subject_type,
CertificateSubjectHandler handler )
{
if ( !handler )
{
logger_->error( "{}: ignored empty certificate handler subject_type={}",
__func__,
subject_type );
return false;
}
auto type_hash = ComputeSubjectTypeHash( subject_type );
if ( type_hash.has_error() )
{
logger_->error( "{}: ignored invalid certificate handler subject_type={}",
__func__,
subject_type );
return false;
}
logger_->debug( "{}: Registering certificate handler subject_type={}",
__func__,
subject_type );
{
std::unique_lock lock( certificate_handlers_mutex_ );
certificate_subject_handlers_[type_hash.value()] = std::move( handler );
}
// A durable certificate may have arrived before its consumer was constructed.
// Recover only after releasing the handler map lock because dispatch reads it.
RecoverPendingCertificateWork();
return true;
}
void ConsensusManager::UnregisterCertificateHandler( std::string_view subject_type )
{
logger_->debug( "{}: Removing Certificate handler with subject_type={}",
__func__,
subject_type );
auto type_hash = ComputeSubjectTypeHash( subject_type );
if ( type_hash.has_error() )
{
return;
}
std::unique_lock lock( certificate_handlers_mutex_ );
certificate_subject_handlers_.erase( type_hash.value() );
}
bool ConsensusManager::RegisterProposalCleanupHandler( std::string_view subject_type,
ProposalCleanupHandler handler )
{
if ( !handler )
{
logger_->error( "{}: ignored empty cleanup handler subject_type={}",
__func__,
subject_type );
return false;
}
auto type_hash = ComputeSubjectTypeHash( subject_type );
if ( type_hash.has_error() )
{
logger_->error( "{}: ignored invalid cleanup handler subject_type={}",
__func__,
subject_type );
return false;
}
logger_->debug( "{}: Registering cleanup handler subject_type={}", __func__, subject_type );
std::unique_lock lock( cleanup_handlers_mutex_ );
proposal_cleanup_handlers_[type_hash.value()].push_back( std::move( handler ) );
return true;
}
void ConsensusManager::UnregisterProposalCleanupHandler( std::string_view subject_type )
{
logger_->debug( "{}: Removing cleanup handler with subject_type={}", __func__, subject_type );
auto type_hash = ComputeSubjectTypeHash( subject_type );
if ( type_hash.has_error() )
{
return;
}
std::unique_lock lock( cleanup_handlers_mutex_ );
proposal_cleanup_handlers_.erase( type_hash.value() );
}
void ConsensusManager::SetPendingLifecycleConfig( PendingLifecycleConfig config )
{
std::lock_guard lock( proposals_mutex_ );
pending_config_ = config;
}
void ConsensusManager::RegisterSlotKeyHandler( std::string_view subject_type, SlotKeyHandler handler )
{
if ( !handler )
{
ConsensusManagerLogger()->error( "{}: ignored empty slot key handler subject_type={}",
__func__,
subject_type );
return;
}
auto type_hash = ComputeSubjectTypeHash( subject_type );
if ( type_hash.has_error() )
{
ConsensusManagerLogger()->error( "{}: ignored invalid slot key handler subject_type={}",
__func__,
subject_type );
return;
}
ConsensusManagerLogger()->debug( "{}: Registering slot key handler subject_type={}", __func__, subject_type );
std::unique_lock lock( slot_key_handlers_mutex_ );
slot_key_handlers_[type_hash.value()] = std::move( handler );
}
void ConsensusManager::UnregisterSlotKeyHandler( std::string_view subject_type )
{
ConsensusManagerLogger()->debug( "{}: Removing slot key handler subject_type={}", __func__, subject_type );
auto type_hash = ComputeSubjectTypeHash( subject_type );
if ( type_hash.has_error() )
{
return;
}
std::unique_lock lock( slot_key_handlers_mutex_ );
slot_key_handlers_.erase( type_hash.value() );
}
void ConsensusManager::FireProposalCleanupCallbacks( const Proposal &proposal )
{
auto subject_hash = GetSubjectHash( proposal.subject() );
if ( subject_hash.has_error() )
{
return;
}
auto nonce_payload = DecodeNonceSubject( proposal.subject() );
if ( nonce_payload.has_error() )
{
return;
}
auto tx_hash = nonce_payload.value().tx_hash();
if ( tx_hash.empty() )
{
return;
}
std::vector<ProposalCleanupHandler> handlers_copy;
{
std::shared_lock lock( cleanup_handlers_mutex_ );
auto it = proposal_cleanup_handlers_.find( proposal.subject().subject_type_hash().hash() );
if ( it != proposal_cleanup_handlers_.end() )
{
handlers_copy = it->second;
}
}
for ( auto &handler : handlers_copy )
{
handler( tx_hash );
}
}
void ConsensusManager::ConfigureTimestampWindow( std::chrono::milliseconds window )
{
if ( window.count() <= 0 )
{
logger_->warn( "{}: using default window", __func__ );
timestamp_window_ms_.store( DEFAULT_TIMESTAMP_WINDOW.count(), std::memory_order_relaxed );
return;
}
timestamp_window_ms_.store( window.count(), std::memory_order_relaxed );
}
void ConsensusManager::ConfigureRoundDuration( std::chrono::milliseconds duration )
{
if ( duration.count() <= 0 )
{
logger_->warn( "{}: using default round duration", __func__ );
round_duration_ms_.store( DEFAULT_ROUND_DURATION.count(), std::memory_order_relaxed );
return;
}
round_duration_ms_.store( duration.count(), std::memory_order_relaxed );
}
void ConsensusManager::ConfigureRoundSkew( std::chrono::milliseconds skew )
{
if ( skew.count() < 0 )
{
logger_->warn( "{}: using default round skew", __func__ );
round_skew_ms_.store( DEFAULT_ROUND_SKEW.count(), std::memory_order_relaxed );
return;
}
round_skew_ms_.store( skew.count(), std::memory_order_relaxed );
}
void ConsensusManager::ConfigureCertificateDelay( std::chrono::milliseconds delay )
{
if ( delay.count() < 0 )
{
logger_->warn( "{}: using zero delay", __func__ );
certificate_delay_ms_.store( 0, std::memory_order_relaxed );
return;
}
certificate_delay_ms_.store( delay.count(), std::memory_order_relaxed );
}
bool ConsensusManager::IsTimestampSane( uint64_t timestamp_ms ) const
{
if ( timestamp_ms == 0 )
{
return false;
}
const auto now_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
std::chrono::system_clock::now().time_since_epoch() )
.count();
const auto window_ms = timestamp_window_ms_.load( std::memory_order_relaxed );
if ( now_ms < 0 || window_ms < 0 )
{
return false;
}
const auto now_u64 = static_cast<uint64_t>( now_ms );
const auto window_u64 = static_cast<uint64_t>( window_ms );
const auto min_ts = ( now_u64 > window_u64 ) ? ( now_u64 - window_u64 ) : 0ULL;
const auto max_ts = ( std::numeric_limits<uint64_t>::max() - now_u64 < window_u64 )
? std::numeric_limits<uint64_t>::max()
: now_u64 + window_u64;
return ( timestamp_ms >= min_ts ) && ( timestamp_ms <= max_ts );
}
uint64_t ConsensusManager::GetCurrentRound( uint64_t proposal_ts_ms ) const
{
if ( proposal_ts_ms == 0 || round_duration_ms_.load( std::memory_order_relaxed ) <= 0 )
{
return 0;
}
const auto now_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
std::chrono::system_clock::now().time_since_epoch() )
.count();
const auto elapsed = static_cast<int64_t>( now_ms ) - static_cast<int64_t>( proposal_ts_ms );
if ( elapsed <= 0 )
{
return 0;
}
const auto skew_ms = round_skew_ms_.load( std::memory_order_relaxed );
if ( elapsed <= skew_ms )
{
return 0;
}
const auto round_ms = round_duration_ms_.load( std::memory_order_relaxed );
auto round = static_cast<uint64_t>( ( elapsed - skew_ms ) / round_ms );
logger_->debug( "{}: Returning round={}", __func__, round );
return round;
}
std::vector<std::string> ConsensusManager::GetOrderedActiveValidators(
const ValidatorRegistry::Registry ®istry ) const
{
std::vector<std::string> validators;
validators.reserve( registry.validators_size() );
for ( const auto &entry : registry.validators() )
{
if ( entry.status() == ValidatorRegistry::Status::ACTIVE )
{
validators.push_back( entry.validator_id() );
}
}
std::sort( validators.begin(), validators.end() );
logger_->trace( "{}: Returning validators with size ={}", __func__, validators.size() );
return validators;
}
ConsensusManager::AggregatorRole ConsensusManager::GetAggregatorRole(
const Proposal &proposal,
const ValidatorRegistry::Registry ®istry ) const
{
logger_->trace( "{}: Checking local aggregator role for proposal", __func__ );
auto ordered = GetOrderedActiveValidators( registry );
if ( ordered.empty() )
{
return AggregatorRole::NotInRegistry;
}
if ( std::find( ordered.begin(), ordered.end(), account_address_ ) == ordered.end() )
{
return AggregatorRole::NotInRegistry;
}
auto hash = sgns::crypto::sha2_256( proposal.proposal_id().data(), proposal.proposal_id().size() );
uint64_t base_index = 0;
for ( size_t i = 0; i < sizeof( uint64_t ) && i < hash.size(); ++i )
{
base_index = ( base_index << 8 ) | hash[i];
}
base_index = base_index % ordered.size();
const auto round = GetCurrentRound( proposal.timestamp() );
const auto index = ( base_index + round ) % ordered.size();
return ordered[index] == account_address_ ? AggregatorRole::CurrentAggregator
: AggregatorRole::ActiveButNotAggregator;
}
outcome::result<std::string> ConsensusManager::GetSubjectHash( const Subject &subject )
{
if ( SubjectTypeMatches( subject, NONCE_SUBJECT_TYPE ) )
{
auto payload = DecodeNonceSubject( subject );
if ( payload.has_error() || payload.value().tx_hash().empty() )
{
return outcome::failure( std::errc::invalid_argument );
}
return payload.value().tx_hash();
}
if ( SubjectTypeMatches( subject, TASK_RESULT_SUBJECT_TYPE ) )
{
auto payload = DecodeTaskResultSubject( subject );
if ( payload.has_error() || payload.value().task_result_hash().empty() )
{
return outcome::failure( std::errc::invalid_argument );
}
return payload.value().task_result_hash();
}
if ( SubjectTypeMatches( subject, REGISTRY_BATCH_SUBJECT_TYPE ) )
{
auto payload = DecodeRegistryBatchSubject( subject );
if ( payload.has_error() || payload.value().batch_root().empty() )
{
return outcome::failure( std::errc::invalid_argument );
}
return std::string( payload.value().batch_root() );
}
return ComputeSubjectId( subject );
}
void ConsensusManager::ContinueProposalAfterSubject( const Proposal &proposal )
{
logger_->debug( "{}: Continuing proposal: hash {}, id {}",
__func__,
GetPrintableSubjectHash( proposal.subject() ),
proposal.proposal_id().substr( 0, 8 ) );
const auto slot_key = GetSlotKey( proposal );
bool process_due_work = false;
bool candidate_admitted = false;
logger_->debug( "{}: Slot key acquired: hash {}, id {}, slot key {}",
__func__,
GetPrintableSubjectHash( proposal.subject() ),
proposal.proposal_id().substr( 0, 8 ),
slot_key );
{
std::lock_guard lock( proposals_mutex_ );
if ( proposals_.find( proposal.proposal_id() ) == proposals_.end() )
{
logger_->debug(
"{}: No proposal state found. Creating... : hash {}, id {}, slot key {}",
__func__,
GetPrintableSubjectHash( proposal.subject() ),
proposal.proposal_id().substr( 0, 8 ),
slot_key );
ProposalState state;
state.proposal = proposal;
state.slot_key = slot_key;
proposals_.emplace( proposal.proposal_id(), std::move( state ) );
}
auto &slot_state = slot_states_[slot_key];
if ( slot_state.candidates_frozen && !slot_state.slot_decided )
{
// Retained for a future attempt, not this one: peers close their local
// candidate windows at different instants, so a contender that is late
// here was in time elsewhere. Dropping it is what let nodes freeze
// different winners for one slot and deadlock it. A decided slot is
// excluded — nothing will re-arbitrate it, so retaining would only copy
// proposals until slot cleanup.
AddCandidateIfAbsent( slot_state.eligible_candidates, proposal );
}
if ( slot_state.active_vote_locked )
{
return;
}
const auto now = std::chrono::steady_clock::now();
auto accepted_certificate = HasAcceptedCertificateForSlot( slot_key );
if ( accepted_certificate.has_error() )
{
// An incomplete legacy scan is never evidence that a slot is unfinalized.
slot_state.certificate_scan_pending = true;
if ( ( slot_state.candidate_deadline == std::chrono::steady_clock::time_point{} ||
now < slot_state.candidate_deadline ) &&
std::none_of( slot_state.scan_pending_candidates.begin(),
slot_state.scan_pending_candidates.end(),
[&proposal]( const ScanPendingCandidate &candidate )
{ return candidate.proposal.proposal_id() == proposal.proposal_id(); } ) )
{
slot_state.scan_pending_candidates.push_back( { proposal, now } );
}
return;
}
slot_state.certificate_scan_pending = false;
if ( accepted_certificate.value() )
{
// Existing accepted legacy data is only a local no-revote fence in Phase 9.
slot_state.active_vote_locked = true;
slot_state.candidates_frozen = true;
slot_state.slot_decided = true;
return;
}
if ( slot_state.candidate_deadline == std::chrono::steady_clock::time_point{} )
{
slot_state.candidate_deadline = now + candidate_window_;
}
if ( now < slot_state.candidate_deadline )
{
for ( const auto &pending_candidate : slot_state.scan_pending_candidates )
{
if ( pending_candidate.admitted_at < slot_state.candidate_deadline )
{
AddCandidateIfAbsent( slot_state.eligible_candidates, pending_candidate.proposal );
}
}
}
slot_state.scan_pending_candidates.clear();
if ( now >= slot_state.candidate_deadline )
{
process_due_work = true;
}
else if ( !slot_state.candidates_frozen )
{
candidate_admitted = AddCandidateIfAbsent( slot_state.eligible_candidates, proposal );
}
if ( candidate_admitted && slot_state.best_proposal_id.empty() )
{
logger_->debug( "{}: Configuring best proposal for hash {}, id={}, slot key {}",
__func__,
GetPrintableSubjectHash( proposal.subject() ),
proposal.proposal_id().substr( 0, 8 ),
slot_key );
slot_state.best_proposal_id = proposal.proposal_id();
auto nonce_payload = DecodeNonceSubject( proposal.subject() );
if ( nonce_payload.has_value() )
{
slot_state.best_tx_hash = nonce_payload.value().tx_hash();
}
}
else if ( candidate_admitted )
{
const auto ¤t = proposals_.at( slot_state.best_proposal_id ).proposal;
logger_->debug(
"{}: Already have a best proposal for hash {}, id={}, slot key {}. Seeing if {} is better ",
__func__,
GetPrintableSubjectHash( current.subject() ),
current.proposal_id().substr( 0, 8 ),
slot_key,
proposal.proposal_id().substr( 0, 8 ) );
if ( IsBetterProposal( proposal, current ) )
{
logger_->debug( "{}: Better proposal for hash {}, id={}, slot key {}. ",
__func__,
GetPrintableSubjectHash( proposal.subject() ),
proposal.proposal_id().substr( 0, 8 ),
slot_key );
slot_state.best_proposal_id = proposal.proposal_id();
auto nonce_payload = DecodeNonceSubject( proposal.subject() );
if ( nonce_payload.has_value() )
{
slot_state.best_tx_hash = nonce_payload.value().tx_hash();
}
}
}
}
auto pending_votes = TakePendingVotes( proposal.proposal_id() );
for ( const auto &vote : pending_votes )
{
HandleVote( vote );
}
if ( process_due_work )
{
ProcessDueVoteWork();
}
}
bool ConsensusManager::CanAdmitPendingProposalLocked( const Proposal &proposal,
std::size_t retained_bytes,
const std::string &proposer_id ) const
{
if ( pending_entries_.size() >= pending_config_.max_pending_proposals )
{
logger_->warn( "{}: pending admission refused: global limit reached proposal_id={}",
__func__,
proposal.proposal_id().substr( 0, 8 ) );
return false;
}
auto proposer_it = pending_count_by_proposer_.find( proposer_id );
if ( proposer_it != pending_count_by_proposer_.end() &&
proposer_it->second >= pending_config_.max_pending_per_proposer )
{
logger_->warn(
"{}: pending admission refused: proposer limit reached proposer={} proposal_id={}",
__func__,
proposer_id.substr( 0, 8 ),
proposal.proposal_id().substr( 0, 8 ) );
return false;
}
if ( pending_retained_bytes_ + retained_bytes > pending_config_.max_retained_pending_bytes )
{
logger_->warn( "{}: pending admission refused: retained byte limit reached proposal_id={}",
__func__,
proposal.proposal_id().substr( 0, 8 ) );
return false;
}
return true;
}
std::vector<ConsensusManager::PendingDependencyKey> ConsensusManager::NormalizePendingDependencies(
const std::string &subject_hash,
const ValidationResult &validation_result ) const
{
if ( !validation_result.dependencies.empty() )
{
return validation_result.dependencies;
}
return { PendingDependencyKey::Certificate( subject_hash ) };
}
std::chrono::milliseconds ConsensusManager::NextPendingRetryDelayLocked( const PendingProposalEntry &entry ) const
{
if ( entry.retry_after.has_value() )
{
return entry.retry_after.value();
}
if ( pending_config_.scheduled_retry_delays.empty() )
{
return std::chrono::seconds( 10 );
}
const auto index = std::min( entry.scheduled_retry_count, pending_config_.scheduled_retry_delays.size() - 1 );
return pending_config_.scheduled_retry_delays[index];
}
bool ConsensusManager::AddPendingProposal( const Proposal &proposal,
const std::string &subject_hash,
const ValidationResult &validation_result,
std::size_t scheduled_retry_count,
std::chrono::steady_clock::time_point last_retry_at )
{
std::lock_guard lock( proposals_mutex_ );
if ( pending_entries_.find( proposal.proposal_id() ) != pending_entries_.end() )
{
RemovePendingProposalLocked( proposal.proposal_id(), "replace" );
}
const auto dependencies = NormalizePendingDependencies( subject_hash, validation_result );
const auto retained_bytes = static_cast<std::size_t>( proposal.ByteSizeLong() );
const auto &proposer_id = proposal.proposer_id();
if ( !CanAdmitPendingProposalLocked( proposal, retained_bytes, proposer_id ) )
{
return false;
}
logger_->debug( "{}: Adding pending proposal for {}: proposal with id {}",
__func__,
subject_hash.substr( 0, 8 ),
proposal.proposal_id().substr( 0, 8 ) );
const auto now = std::chrono::steady_clock::now();
PendingProposalEntry entry;
entry.proposal = proposal;
entry.dependencies = dependencies;
entry.admitted_at = now;
entry.expires_at = now + pending_config_.pending_ttl;
entry.last_retry_at = last_retry_at;
entry.retry_after = validation_result.retry_after;
entry.retained_bytes = retained_bytes;
entry.proposer_id = proposer_id;
entry.scheduled_retry_count = scheduled_retry_count;
entry.next_retry_at = now + NextPendingRetryDelayLocked( entry );
pending_retained_bytes_ += retained_bytes;
pending_count_by_proposer_[proposer_id] += 1;
for ( const auto &dependency : dependencies )
{
pending_by_dependency_[dependency].insert( proposal.proposal_id() );
}
pending_entries_.emplace( proposal.proposal_id(), std::move( entry ) );
timer_cv_.notify_all();
return true;
}
std::vector<ConsensusManager::Proposal> ConsensusManager::TakePendingProposals( const std::string &subject_hash )
{
std::vector<Proposal> result;
std::lock_guard lock( proposals_mutex_ );
const auto dependency = PendingDependencyKey::Certificate( subject_hash );
auto it = pending_by_dependency_.find( dependency );
if ( it == pending_by_dependency_.end() )
{
logger_->trace( "{}: No pending proposals for {}", __func__, subject_hash.substr( 0, 8 ) );
return result;
}
const std::vector<std::string> proposal_ids( it->second.begin(), it->second.end() );
for ( const auto &proposal_id : proposal_ids )
{
auto prop_it = pending_entries_.find( proposal_id );
if ( prop_it != pending_entries_.end() )
{
result.push_back( prop_it->second.proposal );
RemovePendingProposalLocked( proposal_id, "take" );
}
}
logger_->debug( "{}: Taking pending proposals for {}", __func__, subject_hash.substr( 0, 8 ) );
return result;
}
bool ConsensusManager::RemovePendingProposal( const std::string &proposal_id, std::string_view reason )
{
std::lock_guard lock( proposals_mutex_ );
return RemovePendingProposalLocked( proposal_id, reason );
}
bool ConsensusManager::RemovePendingProposalLocked( const std::string &proposal_id, std::string_view reason )
{
auto entry_it = pending_entries_.find( proposal_id );
if ( entry_it == pending_entries_.end() )
{
pending_votes_.erase( proposal_id );
return false;
}
const auto retained_bytes = entry_it->second.retained_bytes;
if ( pending_retained_bytes_ >= retained_bytes )
{
pending_retained_bytes_ -= retained_bytes;
}
else
{
pending_retained_bytes_ = 0;
}
auto proposer_it = pending_count_by_proposer_.find( entry_it->second.proposer_id );
if ( proposer_it != pending_count_by_proposer_.end() )
{
if ( proposer_it->second > 1 )
{
--proposer_it->second;
}
else
{
pending_count_by_proposer_.erase( proposer_it );
}
}
for ( const auto &dependency : entry_it->second.dependencies )
{
auto dep_it = pending_by_dependency_.find( dependency );
if ( dep_it != pending_by_dependency_.end() )
{
dep_it->second.erase( proposal_id );
if ( dep_it->second.empty() )
{
pending_by_dependency_.erase( dep_it );
}
}
}
pending_entries_.erase( entry_it );
pending_votes_.erase( proposal_id );
logger_->debug( "{}: removed pending proposal_id={} reason={}",
__func__,
proposal_id.substr( 0, 8 ),
reason );
return true;
}
void ConsensusManager::RetryPendingProposal( const Proposal &proposal,
std::string_view reason,
std::size_t scheduled_retry_count,
std::chrono::steady_clock::time_point last_retry_at )
{
SubjectHandler subject_handler;
{
std::shared_lock lock( subject_handlers_mutex_ );
auto handler_it = subject_handlers_.find( proposal.subject().subject_type_hash().hash() );
if ( handler_it == subject_handlers_.end() )
{
logger_->error(
"{}: rejected: subject handler missing type_hash={} reason={}",
__func__,
base::hex_lower( gsl::span<const uint8_t>(
reinterpret_cast<const uint8_t *>( proposal.subject().subject_type_hash().hash().data() ),
proposal.subject().subject_type_hash().hash().size() ) ),
reason );
return;
}
subject_handler = handler_it->second;
}
auto subject_result = subject_handler( proposal.subject() );
if ( subject_result.has_error() )
{
logger_->error( "{}: rejected: subject handler error proposal_id={} reason={}",
__func__,
proposal.proposal_id().substr( 0, 8 ),
reason );
return;
}
const auto &validation_result = subject_result.value();
if ( validation_result.check == Check::Reject )
{
logger_->error( "{}: rejected: subject check failed proposal_id={} reason={}",
__func__,
proposal.proposal_id().substr( 0, 8 ),
reason );
return;
}
if ( validation_result.check == Check::Stalled )
{
logger_->warn( "{}: stalled: subject handler stalled proposal_id={} reason={}",
__func__,
proposal.proposal_id().substr( 0, 8 ),
reason );
return;
}
if ( validation_result.check == Check::Pending )
{
auto subject_hash_result = GetSubjectHash( proposal.subject() );
if ( subject_hash_result.has_error() )
{
logger_->error( "{}: rejected: subject hash missing proposal_id={} reason={}",
__func__,
proposal.proposal_id().substr( 0, 8 ),
reason );
return;
}
AddPendingProposal( proposal,
subject_hash_result.value(),
validation_result,
scheduled_retry_count,
last_retry_at );
return;
}
ContinueProposalAfterSubject( proposal );
}
outcome::result<void> ConsensusManager::WakePendingDependency( const PendingDependencyKey &dependency )
{
struct DependencyRetryCandidate
{
Proposal proposal;
std::size_t scheduled_retry_count = 0;
};
std::vector<DependencyRetryCandidate> retry_now;
const auto now = std::chrono::steady_clock::now();
{
std::lock_guard lock( proposals_mutex_ );
auto dep_it = pending_by_dependency_.find( dependency );
if ( dep_it == pending_by_dependency_.end() )
{
return outcome::success();
}
const std::vector<std::string> proposal_ids( dep_it->second.begin(), dep_it->second.end() );
for ( const auto &proposal_id : proposal_ids )
{
auto entry_it = pending_entries_.find( proposal_id );
if ( entry_it == pending_entries_.end() )
{
continue;
}
if ( now >= entry_it->second.expires_at )
{
continue;
}
if ( entry_it->second.last_retry_at != std::chrono::steady_clock::time_point{} &&
now - entry_it->second.last_retry_at < pending_config_.min_dependency_retry_interval )
{
entry_it->second.next_retry_at = entry_it->second.last_retry_at +
pending_config_.min_dependency_retry_interval;
continue;
}
DependencyRetryCandidate candidate;
candidate.proposal = entry_it->second.proposal;
candidate.scheduled_retry_count = entry_it->second.scheduled_retry_count;
entry_it->second.last_retry_at = now;
retry_now.push_back( std::move( candidate ) );
RemovePendingProposalLocked( proposal_id, "dependency-wake" );
}
}
for ( const auto &candidate : retry_now )
{
RetryPendingProposal( candidate.proposal, "dependency-wake", candidate.scheduled_retry_count, now );
}
return outcome::success();
}
void ConsensusManager::ProcessDuePendingRetries()
{
struct RetryCandidate
{
Proposal proposal;
std::size_t scheduled_retry_count = 0;
};
std::vector<RetryCandidate> retry_now;
const auto now = std::chrono::steady_clock::now();
{
std::lock_guard lock( proposals_mutex_ );
for ( auto it = pending_entries_.begin(); it != pending_entries_.end(); )
{
if ( now < it->second.next_retry_at || now >= it->second.expires_at )
{
++it;
continue;
}
const auto proposal_id = it->first;
RetryCandidate candidate;
candidate.proposal = it->second.proposal;
candidate.scheduled_retry_count = it->second.scheduled_retry_count + 1;
it->second.last_retry_at = now;
retry_now.push_back( std::move( candidate ) );
++it;
RemovePendingProposalLocked( proposal_id, "scheduled-retry" );
}
}
for ( const auto &candidate : retry_now )
{
RetryPendingProposal( candidate.proposal, "scheduled-retry", candidate.scheduled_retry_count, now );
}
}
std::string ConsensusManager::ActiveVoteStorageKey( std::string_view slot_key ) const
{
return std::string( ACTIVE_VOTE_BASE_PATH_KEY ) + std::string( slot_key );
}
outcome::result<bool> ConsensusManager::HasAcceptedCertificateForSlot( const std::string &slot_key ) const
{
if ( slot_key.empty() || !db_ )
{
return outcome::failure( std::errc::invalid_argument );
}
if ( fail_accepted_certificate_scan_for_test_ )
{
return outcome::failure( std::errc::io_error );
}
const auto key = std::string( CERTIFICATE_BASE_PATH_KEY ) + slot_key;
auto value = db_->Get( { key } );
if ( value.has_error() )
{
if ( value.error() == storage::DatabaseError::NOT_FOUND )
{
return false;
}
return outcome::failure( value.error() );
}
Certificate certificate;
if ( !certificate.ParseFromArray( value.value().data(), value.value().size() ) ||
!ValidateCertificateKey( certificate, key ) )
{
return outcome::failure( std::errc::invalid_argument );
}
const auto validation = ValidateCertificate( certificate );
if ( validation == Check::Pending || validation == Check::Stalled )
{
return outcome::failure( std::errc::resource_unavailable_try_again );
}
return validation == Check::Approve;
}
outcome::result<bool> ConsensusManager::ReleaseActiveVoteForAcceptedSlot( const std::string &slot_key )
{
{
std::lock_guard lock( fault_test_mutex_ );
++fault_test_counters_.active_vote_release_attempts;
}
auto removed = EraseDurableActiveVoteRecord( slot_key );
if ( removed.has_error() || !removed.value() )
{
return removed;
}
{
std::lock_guard lock( proposals_mutex_ );
active_votes_.erase( slot_key );
}
{
std::lock_guard fault_lock( fault_test_mutex_ );
++fault_test_counters_.active_vote_release_successes;
}
return true;
}
outcome::result<bool> ConsensusManager::EraseDurableActiveVoteRecord( const std::string &slot_key )
{
if ( slot_key.empty() )
{
return outcome::failure( std::errc::invalid_argument );
}
auto datastore = db_ ? db_->GetDataStore() : nullptr;
if ( !datastore )
{
return outcome::failure( std::errc::bad_file_descriptor );
}
crdt::GlobalDB::Buffer key;
key.put( ActiveVoteStorageKey( slot_key ) );
auto existing = datastore->get( key );
if ( existing.has_error() )
{
if ( existing.error() == storage::DatabaseError::NOT_FOUND )
{
return false;
}
return outcome::failure( existing.error() );
}
auto decoded = DecodeActiveVoteRecord( slot_key, existing.value().toString() );
if ( decoded.has_error() )
{
return outcome::failure( decoded.error() );
}
if ( fail_active_vote_removal_for_test_ )
{
return outcome::failure( std::errc::io_error );
}
auto removed = datastore->remove( key );
if ( removed.has_error() )
{
return outcome::failure( removed.error() );
}
return true;
}
bool ConsensusManager::SlotWinnerUnwinnableLocked( const SlotState &slot_state,
const std::string ®istry_cid ) const
{
const auto &winner_id = slot_state.best_proposal_id;
if ( winner_id.empty() )
{
return false;
}
// Cheap first: the healthy case is every voter behind the local winner, and it
// must not pay for a registry load on every timer pass.
if ( std::none_of( slot_state.observed_votes.begin(),
slot_state.observed_votes.end(),
[&winner_id]( const auto &observed )
{ return observed.second.proposal_id() != winner_id; } ) )
{
return false;
}
auto registry_result = registry_->LoadRegistryByCid( registry_cid );
if ( registry_result.has_error() )
{
return false;
}
const auto ®istry = registry_result.value();
const uint64_t total_weight = ValidatorRegistry::TotalWeight( registry );
uint64_t lost_weight = 0;
for ( const auto &[voter_id, observed_vote] : slot_state.observed_votes )
{
if ( observed_vote.proposal_id() != winner_id )
{
lost_weight += ActiveVoterWeight( registry, voter_id );
}
}
if ( lost_weight > total_weight )
{
return false;
}
return !registry_->IsQuorum( total_weight - lost_weight, total_weight );
}
void ConsensusManager::ReopenSlotArbitrationLocked( SlotState &slot_state, const std::string &slot_key )
{
const auto stale_winner_prefix = slot_state.best_proposal_id.substr( 0, 8 );
slot_state.active_vote_locked = false;
slot_state.candidates_frozen = false;
slot_state.best_proposal_id.clear();
slot_state.best_tx_hash.clear();
// eligible_candidates is deliberately kept: IsBetterProposal is a total order on
// proposal content, so arbitrating the full union agrees across nodes.
slot_state.candidate_deadline = std::chrono::steady_clock::now() + candidate_window_;
logger_->info( "{}: re-opened slot arbitration slot={} candidates={} unwinnable_winner={}",
__func__,
slot_key,
slot_state.eligible_candidates.size(),
stale_winner_prefix );
}
void ConsensusManager::CollectUnwinnableSlotsLocked( std::vector<std::string> &slot_keys )
{
// Split-freeze recovery. Peers close their local candidate windows at different
// instants, so they can freeze different winners for one slot and then discard
// each other's votes forever, stranding the slot and every later transaction in
// that account's nonce chain. A winner the votes prove unwinnable cannot be
// certified, so releasing it is safe — unlike release on deadline expiry, which
// must never authorize a replacement
// (CorruptOrExpiredActiveVoteCannotAuthorizeAReplacement).
for ( auto &[slot_key, active_vote] : active_votes_ )
{
auto slot_state_it = slot_states_.find( slot_key );
if ( slot_state_it == slot_states_.end() || slot_state_it->second.slot_decided )
{
continue;
}
auto &slot_state = slot_state_it->second;
// Evidence only changes when a vote disagrees with the local winner, so a
// quiet slot never pays for the registry load SlotWinnerUnwinnableLocked needs.
if ( slot_state.dissent_seen == slot_state.dissent_evaluated )
{
continue;
}
slot_state.dissent_evaluated = slot_state.dissent_seen;
if ( SlotWinnerUnwinnableLocked( slot_state, active_vote.proposal.registry_cid() ) )
{
slot_keys.push_back( slot_key );
}
}
}
void ConsensusManager::ReleaseUnwinnableSlots( const std::vector<std::string> &slot_keys )
{
for ( const auto &slot_key : slot_keys )
{
auto released = EraseDurableActiveVoteRecord( slot_key );
if ( released.has_error() )
{
logger_->error( "{}: cannot release unwinnable slot={}: {}",
__func__,
slot_key,
released.error().message() );
continue;
}
std::lock_guard lock( proposals_mutex_ );
active_votes_.erase( slot_key );
auto slot_state_it = slot_states_.find( slot_key );
if ( slot_state_it != slot_states_.end() )
{
ReopenSlotArbitrationLocked( slot_state_it->second, slot_key );
}
}
}
outcome::result<ActiveVoteRecord> ConsensusManager::BuildActiveVoteRecord( const std::string &slot_key,
const Proposal &proposal,
const Vote &vote,
uint64_t acceptance_deadline_ms ) const
{
std::string proposal_bytes;
std::string vote_bytes;
if ( slot_key.empty() || acceptance_deadline_ms == 0 || !proposal.SerializeToString( &proposal_bytes ) ||
!vote.SerializeToString( &vote_bytes ) )
{
return outcome::failure( std::errc::invalid_argument );
}
ActiveVoteRecord record;
record.set_canonical_slot( slot_key );
record.set_proposal_bytes( proposal_bytes );
record.set_vote_bytes( vote_bytes );
record.set_acceptance_deadline_ms( acceptance_deadline_ms );
return record;
}
outcome::result<ConsensusManager::ActiveVoteState> ConsensusManager::DecodeActiveVoteRecord(
const std::string &slot_key,
std::string_view serialized ) const
{
ActiveVoteRecord record;
if ( slot_key.empty() || serialized.empty() || !record.ParseFromArray( serialized.data(), serialized.size() ) ||
record.canonical_slot() != slot_key || record.proposal_bytes().empty() || record.vote_bytes().empty() ||
record.acceptance_deadline_ms() == 0 )
{
return outcome::failure( std::errc::invalid_argument );
}
ActiveVoteState state;
if ( !state.proposal.ParseFromString( record.proposal_bytes() ) || !state.vote.ParseFromString( record.vote_bytes() ) ||
GetSlotKey( state.proposal ) != slot_key || state.vote.proposal_id() != state.proposal.proposal_id() ||
state.vote.voter_id() != account_address_ || !state.vote.approve() || !CheckVote( state.vote ) )
{
return outcome::failure( std::errc::invalid_argument );
}
auto signing_bytes = VoteSigningBytes( state.vote );
if ( signing_bytes.has_error() ||
!GeniusAccount::VerifySignature( state.vote.voter_id(), state.vote.signature(), signing_bytes.value() ) )
{
return outcome::failure( std::errc::permission_denied );
}
state.acceptance_deadline_ms = record.acceptance_deadline_ms();
state.next_retry_at = std::chrono::steady_clock::now();
return state;
}
outcome::result<ConsensusManager::ActiveVoteState> ConsensusManager::PersistOrLoadExactActiveVote(
const std::string &slot_key,
const Proposal &proposal,
const Vote &vote,
uint64_t acceptance_deadline_ms )
{
auto record = BuildActiveVoteRecord( slot_key, proposal, vote, acceptance_deadline_ms );
if ( record.has_error() || fail_active_vote_persistence_for_test_ )
{
return outcome::failure( record.has_error() ? record.error() : std::make_error_code( std::errc::io_error ) );
}
std::string encoded;
if ( !record.value().SerializeToString( &encoded ) )
{
return outcome::failure( std::errc::invalid_argument );
}
auto datastore = db_ ? db_->GetDataStore() : nullptr;
if ( !datastore )
{
return outcome::failure( std::errc::bad_file_descriptor );
}
crdt::GlobalDB::Buffer key;
key.put( ActiveVoteStorageKey( slot_key ) );
auto existing = datastore->get( key );
if ( existing.has_value() )
{
const std::string existing_bytes( existing.value().toString() );
if ( existing_bytes != encoded )
{
// Exact-vote guard — deliberately NOT relaxed for expiry: the stored
// vote may still be certifying its proposal from other peers'
// retained copies, and peers tally observed votes latest-wins, so an
// expiry-driven replacement could strip weight from a still-live
// winner (CorruptOrExpiredActiveVoteCannotAuthorizeAReplacement).
// Sanctioned replacement goes through
// EraseDurableActiveVoteRecord on provably-safe release instead.
return outcome::failure( std::errc::operation_not_permitted );
}
auto decoded = DecodeActiveVoteRecord( slot_key, existing_bytes );
if ( decoded.has_error() )
{
return outcome::failure( decoded.error() );
}
EnterFinalityFaultBarrier( active_vote_persisted_barrier_ );
return decoded;
}
if ( existing.error() != storage::DatabaseError::NOT_FOUND )
{
return outcome::failure( existing.error() );
}
crdt::GlobalDB::Buffer value;
value.put( encoded );
auto put = datastore->put( key, value );
if ( put.has_error() )
{
return outcome::failure( put.error() );
}
auto decoded = DecodeActiveVoteRecord( slot_key, encoded );
if ( decoded.has_error() )
{
return outcome::failure( decoded.error() );
}
EnterFinalityFaultBarrier( active_vote_persisted_barrier_ );
return decoded;
}
bool ConsensusManager::EnterFinalityFaultBarrier( FinalityFaultBarrier &barrier )
{
std::unique_lock lock( fault_test_mutex_ );
if ( !barrier.armed )
{
return !stop_timer_.load();
}
barrier.entered = true;
fault_test_cv_.notify_all();
(void) fault_test_cv_.wait_for(
lock, std::chrono::seconds( 30 ), [&] { return barrier.released || !barrier.armed || stop_timer_.load(); } );
barrier.entered = false;
return !stop_timer_.load();
}
void ConsensusManager::RecoverActiveVotes()
{
auto datastore = db_ ? db_->GetDataStore() : nullptr;
if ( !datastore )
{
return;
}
crdt::GlobalDB::Buffer prefix;
prefix.put( ACTIVE_VOTE_BASE_PATH_KEY );
auto records = datastore->query( prefix );
if ( records.has_error() )
{
logger_->error( "{}: failed to enumerate durable active votes: {}", __func__, records.error().message() );
return;
}
const auto now_ms = static_cast<uint64_t>( std::chrono::duration_cast<std::chrono::milliseconds>(
std::chrono::system_clock::now().time_since_epoch() )
.count() );
std::lock_guard lock( proposals_mutex_ );
for ( const auto &[key, value] : records.value() )
{
const std::string key_string( key.toString() );
if ( key_string.rfind( std::string( ACTIVE_VOTE_BASE_PATH_KEY ), 0 ) != 0 )
{
continue;
}
const auto slot_key = key_string.substr( ACTIVE_VOTE_BASE_PATH_KEY.size() );
auto decoded = DecodeActiveVoteRecord( slot_key, value.toString() );
if ( decoded.has_error() )
{
logger_->error( "{}: ignored invalid durable active vote slot={}", __func__, slot_key );
continue;
}
// Deadline BEFORE locks: an expired record is a dead attempt — peers
// stop accepting the vote after acceptance_deadline_ms, so re-announcing
// it is pointless — and taking active_vote_locked/candidates_frozen on
// its behalf wedged the slot permanently (no re-vote, no new candidates,
// and nothing to prove unwinnability for a release: the wedge case was
// every restart longer than the ~2s window, i.e. all of them). The
// expired path installs bookkeeping only, so a later attempt can vote
// again under the observed-votes latest-wins tally.
if ( now_ms >= decoded.value().acceptance_deadline_ms )
{
logger_->info( "{}: durable vote for slot {} expired (deadline {} < now {}); "
"recovering bookkeeping only, slot stays votable",
__func__,
slot_key.substr( 0, 8 ),
decoded.value().acceptance_deadline_ms,
now_ms );
auto &slot_state = slot_states_[slot_key];
slot_state.voted_proposal_ids.insert( decoded.value().proposal.proposal_id() );
continue;
}
auto &slot_state = slot_states_[slot_key];
slot_state.active_vote_locked = true;
slot_state.candidates_frozen = true;
slot_state.best_proposal_id = decoded.value().proposal.proposal_id();
slot_state.voted_proposal_ids.insert( decoded.value().proposal.proposal_id() );
auto accepted_certificate = HasAcceptedCertificateForSlot( slot_key );
if ( accepted_certificate.has_error() )
{
slot_state.certificate_scan_pending = true;
if ( now_ms < decoded.value().acceptance_deadline_ms )
{
active_votes_[slot_key] = std::move( decoded.value() );
}
continue;
}
slot_state.certificate_scan_pending = false;
if ( accepted_certificate.value() )
{
continue;
}
if ( now_ms < decoded.value().acceptance_deadline_ms )
{
active_votes_[slot_key] = std::move( decoded.value() );
}
}
}
void ConsensusManager::ProcessDueVoteWork()
{
// Archive nodes are passive replicas: they never emit a self-vote and
// never retry one. The candidate-window bookkeeping below only feeds the
// local self-vote path, so the whole pass is skipped for non-voting roles.
if ( !participates_in_consensus_ )
{
return;
}
std::vector<ActiveVoteState> publish;
std::vector<Vote> replay_votes;
std::vector<std::string> unwinnable_slots;
const auto now_steady = std::chrono::steady_clock::now();
const auto now_ms = static_cast<uint64_t>( std::chrono::duration_cast<std::chrono::milliseconds>(
std::chrono::system_clock::now().time_since_epoch() )
.count() );
{
std::lock_guard lock( proposals_mutex_ );
for ( auto &[slot_key, slot_state] : slot_states_ )
{
if ( slot_state.active_vote_locked || slot_state.candidates_frozen )
{
continue;
}
if ( slot_state.certificate_scan_pending )
{
auto accepted_certificate = HasAcceptedCertificateForSlot( slot_key );
if ( accepted_certificate.has_error() )
{
continue;
}
slot_state.certificate_scan_pending = false;
if ( accepted_certificate.value() )
{
slot_state.active_vote_locked = true;
slot_state.candidates_frozen = true;
continue;
}
if ( slot_state.candidate_deadline == std::chrono::steady_clock::time_point{} )
{
// No window existed while the scan was unavailable. Start the
// fixed window at recovery, retaining every validated contender.
slot_state.candidate_deadline = now_steady + candidate_window_;
}
if ( now_steady < slot_state.candidate_deadline )
{
for ( const auto &pending_candidate : slot_state.scan_pending_candidates )
{
if ( pending_candidate.admitted_at < slot_state.candidate_deadline )
{
AddCandidateIfAbsent( slot_state.eligible_candidates,
pending_candidate.proposal );
}
}
}
// A successful scan after the original deadline deliberately drops
// retained contenders rather than retroactively extending the window.
slot_state.scan_pending_candidates.clear();
}
if ( slot_state.candidate_deadline == std::chrono::steady_clock::time_point{} ||
now_steady < slot_state.candidate_deadline )
{
continue;
}
auto accepted_certificate = HasAcceptedCertificateForSlot( slot_key );
if ( accepted_certificate.has_error() )
{
slot_state.certificate_scan_pending = true;
continue;
}
slot_state.certificate_scan_pending = false;
if ( accepted_certificate.value() )
{
slot_state.active_vote_locked = true;
slot_state.candidates_frozen = true;
slot_state.slot_decided = true;
continue;
}
slot_state.candidates_frozen = true;
if ( slot_state.eligible_candidates.empty() )
{
continue;
}
const Proposal *winner = &slot_state.eligible_candidates.front();
for ( const auto &candidate : slot_state.eligible_candidates )
{
if ( IsBetterProposal( candidate, *winner ) )
{
winner = &candidate;
}
}
slot_state.best_proposal_id = winner->proposal_id();
auto vote = CreateVote( winner->proposal_id(), account_address_, true, signer_, &winner->subject() );
if ( vote.has_error() )
{
continue;
}
const auto deadline_ms = now_ms + static_cast<uint64_t>( candidate_window_.count() );
auto active_vote = PersistOrLoadExactActiveVote( slot_key, *winner, vote.value(), deadline_ms );
if ( active_vote.has_error() )
{
logger_->error( "{}: failed to persist active vote slot={}: {}",
__func__, slot_key, active_vote.error().message() );
continue;
}
if ( stop_timer_.load() )
{
return;
}
slot_state.active_vote_locked = true;
slot_state.voted_proposal_ids.insert( active_vote.value().proposal.proposal_id() );
active_vote.value().next_retry_at = now_steady + active_vote_retry_interval_;
publish.push_back( active_vote.value() );
active_votes_[slot_key] = std::move( active_vote.value() );
// Peers that voted for this winner while it was not the local winner (a
// previous attempt, or a peer that froze first) were recorded but never
// tallied; re-tally them now so the weight already gathered counts.
for ( const auto &[voter_id, observed_vote] : slot_state.observed_votes )
{
if ( observed_vote.proposal_id() == slot_state.best_proposal_id )
{
replay_votes.push_back( observed_vote );
}
}
}
for ( auto &[slot_key, active_vote] : active_votes_ )
{
auto accepted_certificate = HasAcceptedCertificateForSlot( slot_key );
auto slot_state_it = slot_states_.find( slot_key );
if ( accepted_certificate.has_error() )
{
if ( slot_state_it != slot_states_.end() )
{
slot_state_it->second.certificate_scan_pending = true;
}
continue;
}
if ( slot_state_it != slot_states_.end() )
{
slot_state_it->second.certificate_scan_pending = false;
}
if ( accepted_certificate.value() )
{
continue;
}
if ( now_ms >= active_vote.acceptance_deadline_ms || now_steady < active_vote.next_retry_at )
{
// An expired deadline deliberately does NOT release the vote; recovery
// from a split freeze runs in CollectUnwinnableSlotsLocked instead.
continue;
}
active_vote.next_retry_at = now_steady + active_vote_retry_interval_;
publish.push_back( active_vote );
}
CollectUnwinnableSlotsLocked( unwinnable_slots );
}
ReleaseUnwinnableSlots( unwinnable_slots );
for ( const auto &active_vote : publish )
{
if ( stop_timer_.load() )
{
return;
}
std::string bytes;
if ( !active_vote.vote.SerializeToString( &bytes ) )
{
continue;
}
{
std::lock_guard lock( fault_test_mutex_ );
// Friend-scoped observation buffer: bounded so long-running nodes never
// accumulate replayed/retried announcements indefinitely.
constexpr std::size_t kMaxActiveVoteAnnouncementsForTest = 1024;
if ( active_vote_announcements_for_test_.size() >= kMaxActiveVoteAnnouncementsForTest )
{
active_vote_announcements_for_test_.erase(
active_vote_announcements_for_test_.begin(),
active_vote_announcements_for_test_.begin() + kMaxActiveVoteAnnouncementsForTest / 2 );
}
active_vote_announcements_for_test_.push_back( bytes );
}
auto submit_result = SubmitVote( active_vote.vote );
if ( submit_result.has_value() )
{
std::lock_guard lock( fault_test_mutex_ );
++fault_test_counters_.vote_publications;
}
}
for ( const auto &vote : replay_votes )
{
if ( stop_timer_.load() )
{
return;
}
HandleVote( vote );
}
}
void ConsensusManager::ExpirePendingProposals()
{
std::vector<Proposal> expired;
const auto now = std::chrono::steady_clock::now();
{
std::lock_guard lock( proposals_mutex_ );
for ( auto it = pending_entries_.begin(); it != pending_entries_.end(); )
{
if ( now < it->second.expires_at )
{
++it;
continue;
}
const auto proposal_id = it->first;
expired.push_back( it->second.proposal );
++it;
RemovePendingProposalLocked( proposal_id, "ttl-expired" );
}
}
for ( const auto &proposal : expired )
{
FireProposalCleanupCallbacks( proposal );
// Remove ONLY this proposal's entry: ClearProposalSlot erases every
// sibling sharing the canonical slot plus the whole slot state, so a
// TTL expiry on one deferred dependent took a quorate sibling's votes
// and window with it. Full-slot clearing is reserved for slot-terminal
// events (certificate accepted / submitted / slot decided).
std::lock_guard lock( proposals_mutex_ );
proposals_.erase( proposal.proposal_id() );
}
}
void ConsensusManager::AddPendingVote( const Vote &vote )
{
std::lock_guard lock( proposals_mutex_ );
pending_votes_[vote.proposal_id()].push_back( vote );
// Bound the orphan accumulation: a peer replaying votes for proposals
// this node never handled grew one vector per unseen proposal id
// forever (nothing else removed them). Once the map exceeds the bound,
// evict queues whose proposal is unknown and whose newest vote is older
// than the TTL — such a queue can no longer grow or ever be consumed.
if ( pending_votes_.size() <= kMaxTrackedPendingVoteQueues )
{
return;
}
const auto now_ms = static_cast<uint64_t>(
std::chrono::duration_cast<std::chrono::milliseconds>(
std::chrono::system_clock::now().time_since_epoch() )
.count() );
for ( auto it = pending_votes_.begin(); it != pending_votes_.end(); )
{
const bool known = proposals_.find( it->first ) != proposals_.end() ||
pending_entries_.find( it->first ) != pending_entries_.end();
uint64_t newest_ms = 0;
for ( const auto &vote : it->second )
{
newest_ms = std::max<uint64_t>( newest_ms, vote.timestamp() );
}
const bool expired = newest_ms + kPendingVoteQueueTTL.count() < now_ms;
if ( !known && expired )
{
it = pending_votes_.erase( it );
}
else
{
++it;
}
}
}
std::vector<ConsensusManager::Vote> ConsensusManager::TakePendingVotes( const std::string &proposal_id )
{
std::vector<Vote> result;
std::lock_guard lock( proposals_mutex_ );
auto it = pending_votes_.find( proposal_id );
if ( it == pending_votes_.end() )
{
return result;
}
result = std::move( it->second );
pending_votes_.erase( it );
return result;
}
outcome::result<ConsensusManager::Proposal> ConsensusManager::CreateProposal( const Subject &subject,
const std::string &proposer_id,
const std::string ®istry_cid,
uint64_t registry_epoch )
{
return CreateProposal( subject, proposer_id, registry_cid, registry_epoch, signer_ );
}
outcome::result<ConsensusManager::Proposal> ConsensusManager::CreateProposal( const Subject &subject,
const std::string &proposer_id,
const std::string ®istry_cid,
uint64_t registry_epoch,
Signer sign )
{
ConsensusManagerLogger()->trace( "{}: called by {} with hash {}, registry CID {} and epoch {}",
__func__,
proposer_id.substr( 0, 8 ),
GetPrintableSubjectHash( subject ),
registry_cid,
registry_epoch );
if ( !sign )
{
ConsensusManagerLogger()->error( "{}: failed for hash {}: signer is empty",
__func__,
GetPrintableSubjectHash( subject ) );
return outcome::failure( std::errc::invalid_argument );
}
if ( !ValidateSubject( subject ) )
{
ConsensusManagerLogger()->error( "{}: failed for hash {}: subject validation failed",
__func__,
GetPrintableSubjectHash( subject ) );
return outcome::failure( std::errc::invalid_argument );
}
Proposal proposal;
*proposal.mutable_subject() = subject;
proposal.set_proposer_id( proposer_id );
proposal.set_registry_cid( registry_cid );
proposal.set_registry_epoch( registry_epoch );
proposal.set_timestamp(
std::chrono::duration_cast<std::chrono::milliseconds>( std::chrono::system_clock::now().time_since_epoch() )
.count() );
proposal.set_proposal_id( CreateProposalId( proposal ) );
auto signing_bytes = ProposalSigningBytes( proposal );
if ( signing_bytes.has_error() )
{
ConsensusManagerLogger()->error( "{}: failed: signing bytes error={}",
__func__,
signing_bytes.error().message() );
return outcome::failure( signing_bytes.error() );
}
ConsensusManagerLogger()->debug( "{}: Creating proposal ID {} for hash {}",
__func__,
proposal.proposal_id().substr( 0, 8 ),
GetPrintableSubjectHash( subject ) );
BOOST_OUTCOME_TRY( auto &&signature, sign( signing_bytes.value() ) );
proposal.set_signature( signature.data(), signature.size() );
ConsensusManagerLogger()->debug( "{}: success for hash {} proposal_id={}",
__func__,
GetPrintableSubjectHash( subject ),
proposal.proposal_id().substr( 0, 8 ) );
return proposal;
}
outcome::result<ConsensusManager::Vote> ConsensusManager::CreateVote( const std::string &proposal_id,
const std::string &voter_id,
bool approve,
Signer sign,
const Subject *subject )
{
logger_->trace( "{}: called by {}: proposal_id={} approve={}",
__func__,
voter_id.substr( 0, 8 ),
proposal_id.substr( 0, 8 ),
approve );
if ( !sign )
{
logger_->error( "{}: failed: signer is empty", __func__ );
return outcome::failure( std::errc::invalid_argument );
}
Vote vote;
vote.set_proposal_id( proposal_id );
vote.set_voter_id( voter_id );
vote.set_approve( approve );
vote.set_timestamp(
std::chrono::duration_cast<std::chrono::milliseconds>( std::chrono::system_clock::now().time_since_epoch() )
.count() );
// Phase 6 (D-01): populate slot_N_hash fields before signing so the
// signature commits to them (T-06-01). No-op when no populator is set.
// #364: the subject is required — slot hashes must come from the evidence
// gathered while verifying THIS claim, never from endpoint configuration.
// Without a subject there is nothing to bind to, so the vote abstains from
// every slot (fail closed).
SlotHashPopulator slot_hash_populator;
{
std::lock_guard<std::mutex> lock( slot_hash_populator_mutex_ );
slot_hash_populator = slot_hash_populator_;
}
if ( slot_hash_populator && subject != nullptr )
{
slot_hash_populator( vote, *subject );
logger_->debug( "{}: populated slot hashes for proposal_id={}",
__func__,
proposal_id.substr( 0, 8 ) );
}
else if ( slot_hash_populator )
{
logger_->debug( "{}: no subject supplied; abstaining from all RPC slots "
"for proposal_id={}",
__func__,
proposal_id.substr( 0, 8 ) );
}
auto signing_bytes = VoteSigningBytes( vote );
if ( signing_bytes.has_error() )
{
logger_->error( "{}: failed: signing bytes error={}",
__func__,
signing_bytes.error().message() );
return outcome::failure( signing_bytes.error() );
}
BOOST_OUTCOME_TRY( auto &&signature, sign( signing_bytes.value() ) );
vote.set_signature( signature.data(), signature.size() );
logger_->debug( "{}: {} voted for proposal_id={}",
__func__,
voter_id.substr( 0, 8 ),
proposal_id.substr( 0, 8 ) );
return vote;
}
outcome::result<ConsensusManager::VoteBundle> ConsensusManager::CreateVoteBundle( const std::string &proposal_id,
const std::string &aggregator_id,
const std::vector<Vote> &votes,
Signer sign )
{
logger_->trace( "{}: called by {}: proposal_id={} votes={}",
__func__,
aggregator_id.substr( 0, 8 ),
proposal_id.substr( 0, 8 ),
votes.size() );
if ( !sign )
{
logger_->error( "{}: failed: signer is empty", __func__ );
return outcome::failure( std::errc::invalid_argument );
}
VoteBundle bundle;
bundle.set_proposal_id( proposal_id );
bundle.set_aggregator_id( aggregator_id );
bundle.set_timestamp(
std::chrono::duration_cast<std::chrono::milliseconds>( std::chrono::system_clock::now().time_since_epoch() )
.count() );
for ( const auto &vote : votes )
{
*bundle.add_votes() = vote;
}
auto signing_bytes = VoteBundleSigningBytes( bundle );
if ( signing_bytes.has_error() )
{
logger_->error( "{}: failed: signing bytes error={}",
__func__,
signing_bytes.error().message() );
return outcome::failure( signing_bytes.error() );
}
BOOST_OUTCOME_TRY( auto &&signature, sign( signing_bytes.value() ) );
bundle.set_signature( signature.data(), signature.size() );
logger_->debug(
"{}: Vote bundle created successfully by {}: proposal_id={} number of votes={}",
__func__,
aggregator_id.substr( 0, 8 ),
proposal_id.substr( 0, 8 ),
votes.size() );
return bundle;
}
outcome::result<ConsensusManager::Certificate> ConsensusManager::CreateCertificate( const Proposal &proposal,
const std::vector<Vote> &votes )
{
logger_->trace(
"{}: Creating certificate for hash {}: proposal_id={} number of votes={} registry CID={}, epoch={}",
__func__,
GetPrintableSubjectHash( proposal.subject() ),
proposal.proposal_id().substr( 0, 8 ),
votes.size(),
proposal.registry_cid(),
proposal.registry_epoch() );
auto tally_result = TallyVotes( proposal, votes );
if ( tally_result.has_error() )
{
logger_->error( "{}: failed: tally error={}", __func__, tally_result.error().message() );
return outcome::failure( tally_result.error() );
}
const auto &tally = tally_result.value();
Certificate cert;
cert.set_proposal_id( proposal.proposal_id() );
cert.set_registry_cid( proposal.registry_cid() );
cert.set_registry_epoch( proposal.registry_epoch() );
cert.set_total_weight( tally.total_weight );
cert.set_approved_weight( tally.approved_weight );
uint64_t max_vote_ts = 0;
for ( const auto &vote : votes )
{
if ( vote.timestamp() > max_vote_ts )
{
max_vote_ts = vote.timestamp();
}
}
if ( max_vote_ts == 0 )
{
max_vote_ts = std::chrono::duration_cast<std::chrono::milliseconds>(
std::chrono::system_clock::now().time_since_epoch() )
.count();
}
cert.set_timestamp( max_vote_ts );
for ( const auto &vote : votes )
{
*cert.add_votes() = vote;
}
*cert.mutable_proposal() = proposal;
logger_->debug( "{}: Success creating certificate for hash {} proposal_id={}",
__func__,
GetPrintableSubjectHash( proposal.subject() ),
proposal.proposal_id().substr( 0, 8 ) );
return cert;
}
outcome::result<ConsensusManager::QuorumTally> ConsensusManager::TallyVotes(
const Proposal &proposal,
const std::vector<Vote> &votes,
const ValidatorRegistry::Registry ®istry,
const std::string ®istry_cid ) const
{
if ( !proposal.registry_cid().empty() && !registry_cid.empty() && proposal.registry_cid() != registry_cid )
{
logger_->error(
"{}: failed: registry cid mismatch hash {}, proposal CID ={} registry CID={}",
__func__,
GetPrintableSubjectHash( proposal.subject() ),
proposal.registry_cid(),
registry_cid );
return outcome::failure( std::errc::invalid_argument );
}
if ( proposal.registry_epoch() != registry.epoch() )
{
logger_->error(
"{}: failed: registry epoch mismatch hash {}, proposal Epoch={} registry Epoch={}",
__func__,
GetPrintableSubjectHash( proposal.subject() ),
proposal.registry_epoch(),
registry.epoch() );
return outcome::failure( std::errc::invalid_argument );
}
uint64_t total_weight = ValidatorRegistry::TotalWeight( registry );
uint64_t approved_weight = 0;
std::unordered_set<std::string> seen;
// Slot-quorum helpers (EvaluateSlotQuorum/SlotEvidenceReputation) resolve
// membership and weight but never verify signatures, so the bridge-mint
// branch below must only ever receive this signature-verified subset —
// feeding them the raw vector let fabricated votes attributed to real
// ACTIVE validators reach bridge-mint quorum with zero valid signatures.
std::vector<Vote> verified_votes;
verified_votes.reserve( votes.size() );
for ( const auto &vote : votes )
{
logger_->trace( "{}: processing vote for hash {}: voter_id={} approve={}",
__func__,
GetPrintableSubjectHash( proposal.subject() ),
vote.voter_id().substr( 0, 8 ),
vote.approve() );
if ( vote.proposal_id() != proposal.proposal_id() )
{
continue;
}
if ( !seen.insert( vote.voter_id() ).second )
{
continue;
}
const auto *validator = ValidatorRegistry::FindValidator( registry, vote.voter_id() );
if ( !validator || validator->status() != ValidatorRegistry::Status::ACTIVE )
{
logger_->debug( "{}: processing vote for hash {}: voter_id={} approve={}",
__func__,
GetPrintableSubjectHash( proposal.subject() ),
vote.voter_id().substr( 0, 8 ),
vote.approve() );
continue;
}
auto signing_bytes = VoteSigningBytes( vote );
if ( signing_bytes.has_error() )
{
continue;
}
if ( !GeniusAccount::VerifySignature( vote.voter_id(), vote.signature(), signing_bytes.value() ) )
{
continue;
}
verified_votes.push_back( vote );
logger_->debug( "{}: Valid voter signature for hash {}: voter_id={} approve={}",
__func__,
GetPrintableSubjectHash( proposal.subject() ),
vote.voter_id().substr( 0, 8 ),
vote.approve() );
if ( vote.approve() )
{
logger_->debug( "{}: Adding weight for hash {}: voter_id={} weight={}",
__func__,
GetPrintableSubjectHash( proposal.subject() ),
vote.voter_id().substr( 0, 8 ),
validator->weight() );
approved_weight += validator->weight();
}
}
QuorumTally tally;
tally.total_weight = total_weight;
tally.approved_weight = approved_weight;
tally.has_quorum = registry_->IsQuorum( approved_weight, total_weight );
// Phase 6 (D-06): bridge-mint subjects route the quorum decision through
// the cumulative slot model (EvaluateSlotQuorum) instead of the
// single-pool weight sum. The signature-verified approved_weight above is
// retained for observability; the slot tally is authoritative for
// has_quorum on bridge mints. TallyVotes (certificate creation) and the
// incremental HandleVote tally agree via this same dispatcher (Pitfall 1).
// Both slot helpers consume verified_votes only: they count membership
// and weight but never signatures, so unverified entries must never
// reach them.
if ( IsBridgeMintSubject( proposal ) )
{
const auto slot_result = registry_->EvaluateSlotQuorum( verified_votes, registry );
tally.approved_weight = slot_result.total_voting_reputation;
tally.qualified_sum = slot_result.qualified_sum;
tally.slot_threshold = slot_result.threshold;
// D-06 governs only while the mesh's slot evidence can structurally
// reach quorum: the maximum attainable qualified_sum is the reputation
// whose votes carry slot hashes. When that ceiling cannot pass the
// threshold the cumulative model is unreachable for this vote set —
// e.g. an evidence-starved mesh where vote slot population silently
// no-ops — and the single-pool result stands; every approve voter has
// still passed the public-chain witness validation in that case.
if ( SlotEvidenceReputation( verified_votes, registry ) > slot_result.threshold )
{
tally.has_quorum = slot_result.has_quorum;
}
}
logger_->debug(
"{}: Votes tallied for hash {} proposal_id={} approved_weight={} total_weight={} quorum={}",
__func__,
GetPrintableSubjectHash( proposal.subject() ),
proposal.proposal_id().substr( 0, 8 ),
approved_weight,
total_weight,
tally.has_quorum );
return tally;
}
outcome::result<ConsensusManager::QuorumTally> ConsensusManager::TallyVotes( const Proposal &proposal,
const std::vector<Vote> &votes ) const
{
logger_->trace(
"{}: Tallying with current registry for hash {}, proposal_id={} number of votes={}",
__func__,
GetPrintableSubjectHash( proposal.subject() ),
proposal.proposal_id().substr( 0, 8 ),
votes.size() );
if ( proposal.registry_cid().empty() )
{
logger_->error( "{}: failed: proposal registry CID is empty", __func__ );
return outcome::failure( std::errc::invalid_argument );
}
auto registry_result = registry_->LoadRegistryByCid( proposal.registry_cid() );
if ( registry_result.has_error() )
{
logger_->error( "{}: failed: registry load error={} cid={}",
__func__,
registry_result.error().message(),
proposal.registry_cid() );
return outcome::failure( registry_result.error() );
}
return TallyVotes( proposal, votes, registry_result.value(), proposal.registry_cid() );
}
bool ConsensusManager::IsBridgeMintSubject( const Proposal &proposal )
{
// Fail-closed (RESEARCH Pattern 2): any decode failure returns false so
// the single-pool IsQuorum path applies. Only a successfully-decoded
// NonceSubject carrying a kMintV2 with public-chain metadata is a
// bridge mint. Test/local chains can still use a registered
// IInputValidator without being forced through RPC slot quorum.
const auto nonce_payload = DecodeNonceSubject( proposal.subject() );
if ( nonce_payload.has_error() )
{
return false;
}
const auto &transaction = nonce_payload.value().transaction();
if ( transaction.transaction_case() != EmbeddedTransaction::kMintV2 )
{
return false;
}
const auto &mint = transaction.mint_v2();
return IsPublicChainMintChainId( mint.chain_id() );
}
outcome::result<ConsensusManager::QuorumTally> ConsensusManager::EvaluateQuorum(
const Proposal &proposal,
const std::vector<Vote> &votes,
const ValidatorRegistry::Registry ®istry ) const
{
QuorumTally tally;
if ( IsBridgeMintSubject( proposal ) )
{
// Bridge-mint subject: cumulative slot tally (D-06). The slot model
// never verifies signatures itself, so only the signature-verified
// subset may enter it.
const auto verified_votes = SignatureVerifiedVotes( proposal, votes );
const auto slot_result = registry_->EvaluateSlotQuorum( verified_votes, registry );
tally.total_weight = ValidatorRegistry::TotalWeight( registry );
tally.approved_weight = slot_result.total_voting_reputation;
tally.has_quorum = slot_result.has_quorum;
tally.qualified_sum = slot_result.qualified_sum;
tally.slot_threshold = slot_result.threshold;
logger_->debug( "{}: bridge-mint slot tally hash {} proposal_id={} qualified_sum={} "
"threshold={} total_voting_rep={} has_quorum={}",
__func__,
GetPrintableSubjectHash( proposal.subject() ),
proposal.proposal_id().substr( 0, 8 ),
slot_result.qualified_sum,
slot_result.threshold,
slot_result.total_voting_reputation,
slot_result.has_quorum );
return tally;
}
// Non-bridge subject: single-pool IsQuorum path. NOTE: unlike
// TallyVotes' own loop, this branch does NOT re-verify vote signatures —
// callers that need signature-checked tallies (certificate validation)
// keep using TallyVotes' verified loop for non-bridge subjects.
const uint64_t total_weight = ValidatorRegistry::TotalWeight( registry );
uint64_t approved_weight = 0;
std::unordered_set<std::string> seen;
for ( const auto &vote : votes )
{
if ( vote.proposal_id() != proposal.proposal_id() )
{
continue;
}
if ( !seen.insert( vote.voter_id() ).second )
{
continue;
}
const auto *validator = ValidatorRegistry::FindValidator( registry, vote.voter_id() );
if ( !validator || validator->status() != ValidatorRegistry::Status::ACTIVE )
{
continue;
}
if ( vote.approve() )
{
approved_weight += validator->weight();
}
}
tally.total_weight = total_weight;
tally.approved_weight = approved_weight;
tally.has_quorum = registry_->IsQuorum( approved_weight, total_weight );
// qualified_sum / slot_threshold remain zero for non-bridge (observability).
return tally;
}
outcome::result<std::vector<uint8_t>> ConsensusManager::ProposalSigningBytes( const Proposal &proposal )
{
ConsensusManagerLogger()->trace( "{}: called for hash {} proposal_id={}",
__func__,
GetPrintableSubjectHash( proposal.subject() ),
proposal.proposal_id().substr( 0, 8 ) );
return sgns::ProposalSigningBytes( proposal );
}
outcome::result<ValidatorRegistry::Registry> ConsensusManager::LoadRegistryByCidCached(
const std::string &cid ) const
{
{
std::lock_guard lock( registry_cache_mutex_ );
auto it = registry_cache_.find( cid );
if ( it != registry_cache_.end() )
{
return it->second;
}
}
auto loaded = registry_->LoadRegistryByCid( cid );
if ( loaded.has_error() )
{
return loaded;
}
{
std::lock_guard lock( registry_cache_mutex_ );
if ( registry_cache_.size() >= kRegistryCacheMaxEntries )
{
registry_cache_.clear();
}
registry_cache_.emplace( cid, loaded.value() );
}
return loaded;
}
outcome::result<std::vector<uint8_t>> ConsensusManager::VoteSigningBytes( const Vote &vote )
{
ConsensusManagerLogger()->trace( "{}: called with voter address {} proposal_id={}",
__func__,
vote.voter_id().substr( 0, 8 ),
vote.proposal_id() );
return sgns::VoteSigningBytes( vote );
}
std::vector<ConsensusManager::Vote> ConsensusManager::SignatureVerifiedVotes(
const Proposal &proposal,
const std::vector<Vote> &votes )
{
std::vector<Vote> verified;
std::unordered_set<std::string> seen;
verified.reserve( votes.size() );
for ( const auto &vote : votes )
{
if ( vote.proposal_id() != proposal.proposal_id() )
{
continue;
}
if ( !seen.insert( vote.voter_id() ).second )
{
continue;
}
auto signing_bytes = VoteSigningBytes( vote );
if ( signing_bytes.has_error() )
{
continue;
}
if ( !GeniusAccount::VerifySignature( vote.voter_id(), vote.signature(), signing_bytes.value() ) )
{
continue;
}
verified.push_back( vote );
}
return verified;
}
outcome::result<std::vector<uint8_t>> ConsensusManager::VoteBundleSigningBytes( const VoteBundle &bundle )
{
ConsensusManagerLogger()->trace( "{}: called proposal_id={} votes={}",
__func__,
bundle.proposal_id().substr( 0, 8 ),
bundle.votes_size() );
return sgns::VoteBundleSigningBytes( bundle );
}
outcome::result<void> ConsensusManager::SubmitProposal( const Proposal &proposal, bool self_vote )
{
logger_->trace( "{}: called for hash {} proposal_id={} self_vote={}",
__func__,
GetPrintableSubjectHash( proposal.subject() ),
proposal.proposal_id().substr( 0, 8 ),
self_vote );
const auto slot_key = GetSlotKey( proposal );
{
std::lock_guard lock( proposals_mutex_ );
auto it = proposals_.find( proposal.proposal_id() );
if ( it == proposals_.end() )
{
logger_->debug( "{}: Creating proposal state for hash {} proposal_id={}",
__func__,
GetPrintableSubjectHash( proposal.subject() ),
proposal.proposal_id().substr( 0, 8 ) );
ProposalState state;
state.proposal = proposal;
state.slot_key = slot_key;
proposals_.emplace( proposal.proposal_id(), std::move( state ) );
}
}
ConsensusMessage message;
*message.mutable_proposal() = proposal;
auto publish_result = Publish( message );
if ( publish_result.has_error() )
{
logger_->error( "{}: failed: publish error={}",
__func__,
publish_result.error().message() );
return publish_result;
}
logger_->debug( "{}: success for hash {} proposal_id={}",
__func__,
GetPrintableSubjectHash( proposal.subject() ),
proposal.proposal_id().substr( 0, 8 ) );
if ( self_vote )
{
HandleProposal( proposal );
}
return outcome::success();
}
outcome::result<void> ConsensusManager::SubmitVote( const Vote &vote, bool self_handle )
{
logger_->trace( "{}: called by {} proposal_id={}",
__func__,
vote.voter_id().substr( 0, 8 ),
vote.proposal_id().substr( 0, 8 ) );
ConsensusMessage message;
*message.mutable_vote() = vote;
auto result = Publish( message );
if ( result.has_error() )
{
logger_->error( "{}: failed: publish error={}", __func__, result.error().message() );
return result;
}
logger_->debug( "{}: success voter_id={} proposal_id={} ",
__func__,
vote.voter_id().substr( 0, 8 ),
vote.proposal_id().substr( 0, 8 ) );
if ( self_handle )
{
HandleVote( vote );
}
return result;
}
outcome::result<void> ConsensusManager::SubmitCertificate( const Certificate &certificate )
{
logger_->trace( "{}: called for hash {} and proposal_id={}",
__func__,
GetPrintableSubjectHash( certificate.proposal().subject() ),
certificate.proposal_id().substr( 0, 8 ) );
if ( ValidateCertificate( certificate ) != Check::Approve )
{
logger_->error( "{}: rejected invalid certificate proposal_id={}",
__func__,
certificate.proposal_id() );
return outcome::failure( std::errc::invalid_argument );
}
std::string serialized;
if ( !certificate.SerializeToString( &serialized ) )
{
logger_->error( "{}: failed: certificate serialize error", __func__ );
return outcome::failure( std::errc::invalid_argument );
}
const auto key = GetExpectedCertificateSlotKey( certificate );
if ( key.empty() )
{
return outcome::failure( std::errc::invalid_argument );
}
crdt::HierarchicalKey cert_key( key );
crdt::GlobalDB::Buffer cert_value;
cert_value.put( serialized );
auto existing = db_->Get( { key } );
if ( existing.has_value() )
{
Certificate existing_certificate;
if ( !existing_certificate.ParseFromArray( existing.value().data(), existing.value().size() ) ||
!ValidateCertificateKey( existing_certificate, key ) ||
ValidateCertificate( existing_certificate ) != Check::Approve )
{
return outcome::failure( std::errc::invalid_argument );
}
if ( SerializedCertificateHash( existing.value().toString() ) < SerializedCertificateHash( serialized ) )
{
return outcome::success();
}
}
else if ( existing.error() != storage::DatabaseError::NOT_FOUND )
{
return outcome::failure( existing.error() );
}
{
std::lock_guard lock( fault_test_mutex_ );
++fault_test_counters_.certificate_write_attempts;
}
auto cert_put = db_->PutConvergentImmutable( cert_key, cert_value, { consensus_datastore_topic_ } );
if ( cert_put.has_error() )
{
logger_->error( "{}: failed: cert put for hash {} error={}",
__func__,
GetPrintableSubjectHash( certificate.proposal().subject() ),
cert_put.error().message() );
return outcome::failure( cert_put.error() );
}
{
std::lock_guard lock( fault_test_mutex_ );
++fault_test_counters_.certificate_write_successes;
}
// The canonical slot record is the ONLY durable record v3.0 writes.
// A secondary /cert/<subject_hash> copy was considered and rejected:
// it is only readable by develop peers that already fail closed on
// v3.0 registry semantics, and it would give a losing transaction of
// a contended slot a durable, quorum-valid record nothing arbitrates
// against (the slot key converges by lowest hash; distinct subjects
// get distinct hash keys with no mutual ordering). No consensus
// version was ever deployed, so no legacy by-hash records exist and
// the by-hash read path is removed outright — the slot record is the
// sole certificate authority.
if ( !EnterFinalityFaultBarrier( certificate_persisted_barrier_ ) )
{
return outcome::failure( std::errc::operation_canceled );
}
ConsensusMessage message;
*message.mutable_certificate() = certificate;
auto result = Publish( message );
if ( result.has_error() )
{
logger_->error( "{}: certificate persisted but notification failed: {}", __func__, result.error().message() );
return result;
}
logger_->debug( "{}: success submitting certificate for {} and proposal_id={}",
__func__,
GetPrintableSubjectHash( certificate.proposal().subject() ),
certificate.proposal_id().substr( 0, 8 ) );
return outcome::success();
}
void ConsensusManager::HandleProposal( const Proposal &proposal )
{
logger_->trace( "{}: called for hash {} proposal_id={}",
__func__,
GetPrintableSubjectHash( proposal.subject() ),
proposal.proposal_id().substr( 0, 8 ) );
if ( !CheckProposal( proposal ) )
{
logger_->error( "{}: rejected: Invalid proposal for hash {} proposal_id={}",
__func__,
GetPrintableSubjectHash( proposal.subject() ),
proposal.proposal_id().substr( 0, 8 ) );
return;
}
if ( !IsTimestampSane( proposal.timestamp() ) )
{
logger_->error( "{}: rejected: timestamp out of bounds for hash {} proposal_id={}",
__func__,
GetPrintableSubjectHash( proposal.subject() ),
proposal.proposal_id().substr( 0, 8 ) );
return;
}
if ( proposal.registry_cid().empty() )
{
logger_->error( "{}: rejected: proposal registry CID missing for hash {}. proposal_id={}",
__func__,
GetPrintableSubjectHash( proposal.subject() ),
proposal.proposal_id().substr( 0, 8 ) );
return;
}
auto subject_hash = GetSubjectHash( proposal.subject() );
if ( subject_hash.has_error() )
{
logger_->error( "{}: rejected: subject hash missing proposal_id={}",
__func__,
proposal.proposal_id().substr( 0, 8 ) );
return;
}
auto proposal_registry_result = registry_->LoadRegistryByCid( proposal.registry_cid() );
if ( proposal_registry_result.has_error() )
{
logger_->warn(
"{}: deferred: registry load error={} proposal={} proposal_id={} hash={}. Keeping proposal pending",
__func__,
proposal_registry_result.error().message(),
proposal.registry_cid(),
proposal.proposal_id().substr( 0, 8 ),
subject_hash.value().substr( 0, 8 ) );
{
std::lock_guard lock( proposals_mutex_ );
if ( proposals_.find( proposal.proposal_id() ) == proposals_.end() )
{
ProposalState state;
state.proposal = proposal;
state.slot_key = GetSlotKey( proposal );
proposals_.emplace( proposal.proposal_id(), std::move( state ) );
}
}
AddPendingProposal( proposal, subject_hash.value() );
return;
}
if ( proposal.registry_epoch() != proposal_registry_result.value().epoch() )
{
logger_->error( "{}: rejected: registry epoch mismatch proposal={} registry={}",
__func__,
proposal.registry_epoch(),
proposal_registry_result.value().epoch() );
return;
}
if ( !CheckSubject( proposal.subject() ) )
{
logger_->error( "{}: rejected: subject check failed for hash {} proposal_id={}",
__func__,
GetPrintableSubjectHash( proposal.subject() ),
proposal.proposal_id().substr( 0, 8 ) );
return;
}
auto accepted_certificate = HasAcceptedCertificateForSlot( GetSlotKey( proposal ) );
if ( accepted_certificate.has_error() )
{
return;
}
if ( accepted_certificate.value() )
{
logger_->debug( "{}: ignored: subject already certified hash={} proposal_id={}",
__func__,
subject_hash.value().substr( 0, 8 ),
proposal.proposal_id().substr( 0, 8 ) );
std::lock_guard lock( proposals_mutex_ );
RemovePendingProposalLocked( proposal.proposal_id(), "already-certified" );
return;
}
SubjectHandler subject_handler;
{
std::shared_lock lock( subject_handlers_mutex_ );
auto handler_it = subject_handlers_.find( proposal.subject().subject_type_hash().hash() );
if ( handler_it == subject_handlers_.end() )
{
logger_->error(
"{}: rejected: subject handler missing type_hash={}",
__func__,
base::hex_lower( gsl::span<const uint8_t>(
reinterpret_cast<const uint8_t *>( proposal.subject().subject_type_hash().hash().data() ),
proposal.subject().subject_type_hash().hash().size() ) ) );
return;
}
subject_handler = handler_it->second;
}
auto subject_result = subject_handler( proposal.subject() );
if ( subject_result.has_error() )
{
logger_->error( "{}: rejected: subject handler error for hash {} proposal_id={}",
__func__,
GetPrintableSubjectHash( proposal.subject() ),
proposal.proposal_id().substr( 0, 8 ) );
return;
}
const auto &validation_result = subject_result.value();
if ( validation_result.check == Check::Reject )
{
logger_->error( "{}: rejected: subject check failed for hash {} proposal_id={}",
__func__,
GetPrintableSubjectHash( proposal.subject() ),
proposal.proposal_id().substr( 0, 8 ) );
return;
}
if ( validation_result.check == Check::Stalled )
{
logger_->warn( "{}: stalled: subject handler stalled for hash {} proposal_id={}",
__func__,
GetPrintableSubjectHash( proposal.subject() ),
proposal.proposal_id().substr( 0, 8 ) );
return;
}
if ( validation_result.check == Check::Pending )
{
{
std::lock_guard lock( proposals_mutex_ );
if ( proposals_.find( proposal.proposal_id() ) == proposals_.end() )
{
ProposalState state;
state.proposal = proposal;
state.slot_key = GetSlotKey( proposal );
proposals_.emplace( proposal.proposal_id(), std::move( state ) );
}
}
logger_->debug( "{}: Adding pending proposal for hash {} proposal_id={}",
__func__,
GetPrintableSubjectHash( proposal.subject() ),
proposal.proposal_id().substr( 0, 8 ) );
AddPendingProposal( proposal, subject_hash.value(), validation_result );
return;
}
ContinueProposalAfterSubject( proposal );
}
outcome::result<void> ConsensusManager::ResumeProposalHandling( const std::string &subject_hash )
{
if ( subject_hash.empty() )
{
return outcome::failure( std::errc::invalid_argument );
}
logger_->trace( "{}: Attempting to resume proposals for hash={}",
__func__,
subject_hash.substr( 0, 8 ) );
auto to_process = TakePendingProposals( subject_hash );
for ( const auto &proposal : to_process )
{
SubjectHandler subject_handler;
{
std::shared_lock lock( subject_handlers_mutex_ );
auto handler_it = subject_handlers_.find( proposal.subject().subject_type_hash().hash() );
if ( handler_it == subject_handlers_.end() )
{
logger_->error(
"{}: rejected: subject handler missing type_hash={}",
__func__,
base::hex_lower( gsl::span<const uint8_t>(
reinterpret_cast<const uint8_t *>( proposal.subject().subject_type_hash().hash().data() ),
proposal.subject().subject_type_hash().hash().size() ) ) );
continue;
}
subject_handler = handler_it->second;
}
auto subject_result = subject_handler( proposal.subject() );
if ( subject_result.has_error() )
{
logger_->error( "{}: rejected: subject handler error for hash {} proposal_id={}",
__func__,
subject_hash.substr( 0, 8 ),
proposal.proposal_id().substr( 0, 8 ) );
continue;
}
const auto &validation_result = subject_result.value();
if ( validation_result.check == Check::Reject )
{
logger_->error( "{}: rejected: subject check failed for hash {} proposal_id={}",
__func__,
subject_hash.substr( 0, 8 ),
proposal.proposal_id().substr( 0, 8 ) );
continue;
}
if ( validation_result.check == Check::Stalled )
{
logger_->warn( "{}: stalled: subject handler stalled for hash {} proposal_id={}",
__func__,
subject_hash.substr( 0, 8 ),
proposal.proposal_id().substr( 0, 8 ) );
continue;
}
if ( validation_result.check == Check::Pending )
{
auto subject_hash_result = GetSubjectHash( proposal.subject() );
if ( subject_hash_result.has_error() )
{
logger_->error( "{}: rejected: subject hash missing proposal_id={}",
__func__,
proposal.proposal_id() );
continue;
}
logger_->debug( "{}: Adding pending proposal for hash {} proposal_id={}",
__func__,
subject_hash.substr( 0, 8 ),
proposal.proposal_id().substr( 0, 8 ) );
AddPendingProposal( proposal, subject_hash_result.value(), validation_result );
continue;
}
ContinueProposalAfterSubject( proposal );
}
return outcome::success();
}
void ConsensusManager::ProcessCertificates()
{
// Defense in depth. Aggregation is otherwise gated only by validator-registry membership,
// which an Archive normally never attains precisely because it never votes. That is a
// side effect, not a guarantee — a genesis-seeded Archive (SetAdditionalGenesisValidator-
// Addresses) is ACTIVE without ever having voted. Refuse by role as well as by registry.
if ( !participates_in_consensus_ )
{
return;
}
std::vector<ProposalState> to_process;
{
std::lock_guard lock( proposals_mutex_ );
for ( auto &kv : proposals_ )
{
auto &state = kv.second;
if ( !state.quorum_reached )
{
logger_->debug(
"{}: Found proposal without quorum reached for hash {} proposal_id={}",
__func__,
GetPrintableSubjectHash( state.proposal.subject() ),
state.proposal.proposal_id().substr( 0, 8 ) );
continue;
}
to_process.push_back( state );
}
}
for ( auto &state : to_process )
{
auto accepted_certificate = HasAcceptedCertificateForSlot( state.slot_key );
if ( accepted_certificate.has_error() )
{
logger_->error( "{}: slot lookup failed, no certificate this tick for hash {} "
"proposal_id={} error={}",
__func__,
GetPrintableSubjectHash( state.proposal.subject() ),
state.proposal.proposal_id().substr( 0, 8 ),
accepted_certificate.error().message() );
continue;
}
if ( accepted_certificate.value() )
{
logger_->debug( "{}: slot {} already certified, clearing proposal_id={}",
__func__,
state.slot_key.substr( 0, 8 ),
state.proposal.proposal_id().substr( 0, 8 ) );
ClearProposalSlot( state.proposal );
continue;
}
logger_->debug( "{}: Processing proposal with quorum reached for hash {} proposal_id={}",
__func__,
GetPrintableSubjectHash( state.proposal.subject() ),
state.proposal.proposal_id().substr( 0, 8 ) );
const auto now_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
std::chrono::system_clock::now().time_since_epoch() )
.count();
if ( state.quorum_reached_ts_ms != 0 && certificate_delay_ms_.load( std::memory_order_relaxed ) > 0 )
{
const auto elapsed_ms = static_cast<int64_t>( now_ms ) -
static_cast<int64_t>( state.quorum_reached_ts_ms );
if ( elapsed_ms < certificate_delay_ms_.load( std::memory_order_relaxed ) )
{
logger_->debug( "{}: certificate delay not elapsed for hash {} proposal_id={} "
"elapsed_ms={} delay_ms={}",
__func__,
GetPrintableSubjectHash( state.proposal.subject() ),
state.proposal.proposal_id().substr( 0, 8 ),
elapsed_ms,
certificate_delay_ms_.load( std::memory_order_relaxed ) );
continue;
}
}
const auto round = GetCurrentRound( state.proposal.timestamp() );
if ( state.last_attempt_round != NO_ROUND && round == state.last_attempt_round )
{
logger_->debug(
"{}: proposal already attempted in round for hash {} proposal_id={} round={}",
__func__,
GetPrintableSubjectHash( state.proposal.subject() ),
state.proposal.proposal_id().substr( 0, 8 ),
round );
continue;
}
auto proposal_registry_result = LoadRegistryByCidCached( state.proposal.registry_cid() );
if ( proposal_registry_result.has_error() )
{
logger_->debug( "{}: skipping proposal due to registry load error={} proposal_id={}",
__func__,
proposal_registry_result.error().message(),
state.proposal.proposal_id().substr( 0, 8 ) );
continue;
}
const auto &proposal_registry = proposal_registry_result.value();
if ( state.proposal.registry_epoch() != proposal_registry.epoch() )
{
logger_->debug( "{}: skipping proposal due to registry epoch mismatch proposal_id={}",
__func__,
state.proposal.proposal_id().substr( 0, 8 ) );
continue;
}
const auto aggregator_role = GetAggregatorRole( state.proposal, proposal_registry );
if ( aggregator_role == AggregatorRole::NotInRegistry )
{
logger_->debug(
"{}: local node not in proposal registry; clearing local proposal for hash {} proposal_id={}",
__func__,
GetPrintableSubjectHash( state.proposal.subject() ),
state.proposal.proposal_id().substr( 0, 8 ) );
ClearProposalSlot( state.proposal );
continue;
}
if ( aggregator_role == AggregatorRole::ActiveButNotAggregator )
{
logger_->debug( "{}: not aggregator for proposal for hash {} proposal_id={}",
__func__,
GetPrintableSubjectHash( state.proposal.subject() ),
state.proposal.proposal_id().substr( 0, 8 ) );
continue;
}
{
std::lock_guard lock( proposals_mutex_ );
auto it = proposals_.find( state.proposal.proposal_id() );
if ( it != proposals_.end() )
{
it->second.last_attempt_round = round;
}
}
// Snapshot the aggregator-decision round for the fault-test barrier:
// GetAggregatorRole just evaluated GetCurrentRound(proposal.timestamp())
// to select this publisher; observers reading the round after the
// persist pause may have crossed into the next round.
{
std::lock_guard lock( fault_test_mutex_ );
if ( certificate_persisted_barrier_.armed )
{
certificate_persisted_barrier_.entered_round = GetCurrentRound( state.proposal.timestamp() );
}
}
logger_->debug( "{}: Attempting to create certificate for hash {} proposal_id={} round={}",
__func__,
GetPrintableSubjectHash( state.proposal.subject() ),
state.proposal.proposal_id().substr( 0, 8 ),
round );
auto certificate_result = CreateCertificate( state.proposal, state.votes );
if ( certificate_result.has_error() )
{
logger_->error(
"{}: failed: certificate creation error for hash {} proposal_id {}: {}",
__func__,
GetPrintableSubjectHash( state.proposal.subject() ),
state.proposal.proposal_id().substr( 0, 8 ),
certificate_result.error().message() );
continue;
}
if ( ValidateCertificate( certificate_result.value() ) != Check::Approve )
{
logger_->error( "{}: rejected invalid generated certificate proposal_id={}",
__func__,
state.proposal.proposal_id() );
continue;
}
auto submit_result = SubmitCertificate( certificate_result.value() );
if ( submit_result.has_error() )
{
logger_->error( "{}: certificate submission failed for hash {} proposal_id={} "
"error={}",
__func__,
GetPrintableSubjectHash( state.proposal.subject() ),
state.proposal.proposal_id().substr( 0, 8 ),
submit_result.error().message() );
continue;
}
ClearProposalSlot( state.proposal );
logger_->debug( "{}: certificate submitted for hash {} proposal_id={}",
__func__,
GetPrintableSubjectHash( state.proposal.subject() ),
state.proposal.proposal_id().substr( 0, 8 ) );
}
}
void ConsensusManager::UpdateCertificatesPending()
{
bool has_pending = false;
{
std::lock_guard lock( proposals_mutex_ );
for ( const auto &kv : proposals_ )
{
if ( kv.second.quorum_reached )
{
has_pending = true;
break;
}
}
}
certificates_pending_.store( has_pending );
if ( !has_pending )
{
timer_cv_.notify_all();
}
}
bool ConsensusManager::RegisterCertificateFilter()
{
const std::string pattern = std::string( CERT_KEY_PATTERN );
auto weak_self = weak_from_this();
// Register the unique-keyed callback first: CRDTCallbackManager rejects
// duplicate patterns, while CRDTDataFilter::RegisterElementFilter silently
// REPLACES an existing entry. Registering the filter first would clobber
// the pattern entry owned by a still-live manager on the same GlobalDB
// (e.g. the certificate-signing manager a test or account switch creates),
// and this manager's Close() would then unregister the survivor's filter.
certificate_callback_registered_ = db_->RegisterNewElementCallback(
pattern,
[weak_self]( crdt::CRDTCallbackManager::NewDataPair new_data, const std::string &cid )
{
if ( auto strong = weak_self.lock() )
{
strong->CertificateReceived( std::move( new_data ), cid );
}
} );
certificate_filter_registered_ = certificate_callback_registered_
&& db_->RegisterElementFilter(
pattern,
[weak_self]( const crdt::pb::Element &element )
{
if ( auto strong = weak_self.lock() )
{
return strong->FilterCertificate( element );
}
return crdt::CRDTDataFilter::ElementFilterResult::Accept();
} );
db_->AddListenTopic( consensus_datastore_topic_ );
return certificate_filter_registered_ && certificate_callback_registered_;
}
crdt::CRDTDataFilter::ElementFilterResult ConsensusManager::FilterCertificate(
const crdt::pb::Element &element )
{
logger_->trace( "{}: entry key={}", __func__, element.key() );
Certificate certificate;
if ( !certificate.ParseFromString( element.value() ) )
{
logger_->error( "{}: parse failed, rejecting: {}", __func__, element.key() );
return crdt::CRDTDataFilter::ElementFilterResult::Reject();
}
if ( certificate.proposal_id().empty() )
{
logger_->error( "{}: missing proposal_id, rejecting: {}", __func__, element.key() );
return crdt::CRDTDataFilter::ElementFilterResult::Reject();
}
if ( !ValidateCertificateKey( certificate, element.key() ) )
{
logger_->error( "{}: slot key binding failed, rejecting: {}", __func__, element.key() );
return crdt::CRDTDataFilter::ElementFilterResult::Reject();
}
const auto validation = ValidateCertificate( certificate );
if ( validation == Check::Reject )
{
logger_->error( "{}: validation rejected, dropping: {}", __func__, element.key() );
return crdt::CRDTDataFilter::ElementFilterResult::Reject();
}
if ( validation == Check::Stalled )
{
// The certificate is exactly key-bound and structurally validated; only
// its registry-dependent quorum is deferred behind a registry snapshot
// this node cannot load yet. Storing it anyway would durably occupy the
// canonical slot with a record that may never validate (an attacker
// needs only a self-signed proposal referencing a nonexistent registry
// CID), permanently blocking overwrites and local voting. Stall the
// whole delta instead: no head is recorded and the failed-root retry
// machinery re-evaluates once the referenced registry snapshot syncs.
logger_->info( "{}: validation stalled, retrying certificate after registry sync: {}",
__func__,
element.key() );
return crdt::CRDTDataFilter::ElementFilterResult::Stall();
}
auto existing = db_->Get( { element.key() } );
if ( existing.has_error() )
{
if ( existing.error() != storage::DatabaseError::NOT_FOUND )
{
logger_->error( "{}: existing certificate read failed, rejecting: {}", __func__, element.key() );
return crdt::CRDTDataFilter::ElementFilterResult::Reject();
}
logger_->debug( "{}: certificate accepted key={}", __func__, element.key() );
return crdt::CRDTDataFilter::ElementFilterResult::Accept();
}
Certificate existing_certificate;
const auto existing_serialized = std::string( existing.value().toString() );
if ( !existing_certificate.ParseFromString( existing_serialized ) ||
!ValidateCertificateKey( existing_certificate, element.key() ) )
{
logger_->error( "{}: invalid existing certificate, rejecting: {}", __func__, element.key() );
return crdt::CRDTDataFilter::ElementFilterResult::Reject();
}
const auto existing_validation = ValidateCertificate( existing_certificate );
if ( existing_validation == Check::Reject )
{
logger_->error( "{}: existing certificate rejected validation, rejecting: {}",
__func__,
element.key() );
return crdt::CRDTDataFilter::ElementFilterResult::Reject();
}
// A stored Approve-or-Stalled value falls through to the unchanged
// lower-hash ordering below.
const auto candidate_mint_hash = VerifiedMintV2TransactionHash( certificate );
const auto existing_mint_hash = VerifiedMintV2TransactionHash( existing_certificate );
if ( candidate_mint_hash && existing_mint_hash && candidate_mint_hash != existing_mint_hash )
{
logger_->critical(
"{}: consensus equivocation canonical_slot={} existing_tx_hash={} candidate_tx_hash={}",
__func__,
element.key(),
*existing_mint_hash,
*candidate_mint_hash );
}
if ( SerializedCertificateHash( existing_serialized ) < SerializedCertificateHash( element.value() ) )
{
logger_->error( "{}: higher serialized certificate hash rejected key={}", __func__, element.key() );
return crdt::CRDTDataFilter::ElementFilterResult::Reject();
}
logger_->debug( "{}: certificate accepted key={}", __func__, element.key() );
return crdt::CRDTDataFilter::ElementFilterResult::Accept();
}
void ConsensusManager::CertificateReceived( crdt::CRDTCallbackManager::NewDataPair new_data,
const std::string &cid )
{
auto [key, value] = new_data;
(void) cid;
(void) value;
// CrdtSet invokes this callback before committing its batch. Receipt therefore
// cannot authorize certificate handling or removal of the local vote lock.
{
std::lock_guard lock( fault_test_mutex_ );
++fault_test_counters_.certificate_notifications_received;
}
certificate_work_journal_->MarkSeen( key );
certificate_work_journal_->MarkStalled( key, std::chrono::milliseconds( 0 ) );
timer_cv_.notify_all();
}
ConsensusManager::Check ConsensusManager::ValidateCertificate( const Certificate &certificate ) const
{
if ( certificate.proposal_id().empty() )
{
logger_->error( "{}: Certificate proposal ID missing ", __func__ );
return Check::Reject;
}
if ( !certificate.has_proposal() )
{
logger_->error( "{}: Certificate missing proposal ", __func__ );
return Check::Reject;
}
const auto &proposal = certificate.proposal();
if ( proposal.proposal_id() != certificate.proposal_id() )
{
logger_->error( "{}: rejected: proposal_id mismatch cert={} proposal={}",
__func__,
certificate.proposal_id(),
proposal.proposal_id() );
return Check::Reject;
}
if ( proposal.registry_cid() != certificate.registry_cid() ||
proposal.registry_epoch() != certificate.registry_epoch() )
{
logger_->error( "{}: rejected: registry mismatch proposal_id={}",
__func__,
certificate.proposal_id() );
return Check::Reject;
}
if ( !ValidateSubject( proposal.subject() ) )
{
logger_->error( "{}: rejected: invalid subject proposal_id={}",
__func__,
proposal.proposal_id() );
return Check::Reject;
}
if ( !CheckProposal( proposal ) )
{
logger_->error( "{}: rejected: invalid proposal proposal_id={}",
__func__,
proposal.proposal_id() );
return Check::Reject;
}
const auto computed_id = CreateProposalId( proposal );
if ( computed_id.empty() )
{
logger_->error( "{}: rejected: computed_id empty", __func__ );
return Check::Reject;
}
if ( computed_id != certificate.proposal_id() )
{
logger_->error( "{}: rejected: computed_id mismatch cert={} computed={}",
__func__,
certificate.proposal_id(),
computed_id );
return Check::Reject;
}
if ( !ValidateCertificateBinding( certificate ) )
{
logger_->error( "{}: rejected: canonical slot binding failed proposal_id={}",
__func__,
certificate.proposal_id() );
return Check::Reject;
}
auto registry_ret = registry_->LoadRegistryByCid( certificate.registry_cid() );
if ( registry_ret.has_error() )
{
logger_->error( "{}: registry load pending error={} for registry cid {} proposal_id={}",
__func__,
registry_ret.error().message(),
certificate.registry_cid(),
certificate.proposal_id() );
return Check::Stalled;
}
auto ®istry = registry_ret.value();
std::vector<Vote> votes;
votes.reserve( static_cast<size_t>( certificate.votes_size() ) );
for ( const auto &vote : certificate.votes() )
{
votes.push_back( vote );
}
auto tally = TallyVotes( proposal, votes, registry, certificate.registry_cid() );
if ( tally.has_error() || !tally.value().has_quorum )
{
logger_->warn(
"{}: rejected: certificate tally failed slot_key={} registry_cid={} votes={} tally_error={}",
__func__,
GetSlotKey( certificate.proposal() ),
certificate.registry_cid(),
certificate.votes_size(),
tally.has_error() ? tally.error().message() : std::string( "quorum not reached" ) );
return Check::Reject;
}
return Check::Approve;
}
bool ConsensusManager::ValidateCertificateBinding( const Certificate &certificate )
{
return certificate.has_proposal() && !GetSlotKey( certificate.proposal() ).empty();
}
bool ConsensusManager::ValidateCertificateKey( const Certificate &certificate, std::string_view key )
{
if ( !ValidateCertificateBinding( certificate ) )
{
return false;
}
if ( key == GetExpectedCertificateSlotKey( certificate ) )
{
return true;
}
// Secondary subject-hash index record (develop consumer contract): the
// canonical slot record above stays authoritative for consensus
// internals; this record only makes the certificate retrievable by its
// subject hash (e.g. a transaction hash).
auto subject_hash = GetSubjectHash( certificate.proposal().subject() );
return subject_hash.has_value() &&
key == std::string{ CERTIFICATE_BASE_PATH_KEY } + subject_hash.value();
}
std::string ConsensusManager::GetExpectedCertificateSlotKey( const Certificate &certificate )
{
if ( !ValidateCertificateBinding( certificate ) )
{
return {};
}
return std::string{ CERTIFICATE_BASE_PATH_KEY } + GetSlotKey( certificate.proposal() );
}
void ConsensusManager::HandleVote( const Vote &vote )
{
logger_->trace( "{}: called. Vote by {} on proposal_id={} ",
__func__,
vote.voter_id().substr( 0, 8 ),
vote.proposal_id().substr( 0, 8 ) );
if ( !CheckVote( vote ) )
{
logger_->error( "{}: rejected: Invalid vote proposal_id={} voter_id={}",
__func__,
vote.proposal_id(),
vote.voter_id() );
return;
}
if ( !vote.approve() )
{
logger_->debug( "{}: ignored: vote not approved voter_id={}",
__func__,
vote.voter_id().substr( 0, 8 ) );
//TODO - maybe see reputation?
return;
}
auto signing_bytes = VoteSigningBytes( vote );
if ( signing_bytes.has_error() )
{
logger_->error( "{}: rejected: signing bytes error={}",
__func__,
signing_bytes.error().message() );
return;
}
if ( !GeniusAccount::VerifySignature( vote.voter_id(), vote.signature(), signing_bytes.value() ) )
{
logger_->error( "{}: rejected: signature verification failed voter_id={}",
__func__,
vote.voter_id().substr( 0, 8 ) );
return;
}
bool has_quorum = false;
{
std::lock_guard lock( proposals_mutex_ );
auto it = proposals_.find( vote.proposal_id() );
if ( it == proposals_.end() )
{
pending_votes_[vote.proposal_id()].push_back( vote );
logger_->debug( "{}: queued pending vote proposal_id={}",
__func__,
vote.proposal_id().substr( 0, 8 ) );
return;
}
auto &proposal_state = it->second;
auto accepted_certificate = HasAcceptedCertificateForSlot( proposal_state.slot_key );
if ( accepted_certificate.has_value() && accepted_certificate.value() )
{
logger_->debug( "{}: ignored: vote for already certified slot {} proposal_id={}",
__func__,
proposal_state.slot_key.substr( 0, 8 ),
vote.proposal_id().substr( 0, 8 ) );
pending_votes_.erase( vote.proposal_id() );
return;
}
auto slot_it = slot_states_.find( proposal_state.slot_key );
if ( slot_it != slot_states_.end() )
{
auto &slot_state = slot_it->second;
// Retained, not dropped: a later attempt may crown this proposal, and the
// peer stops re-sending once its own acceptance deadline passes, so a drop
// is permanent. This is also the evidence ReleaseUnwinnableSlotsLocked
// reasons over — under the one-vote-per-slot rule, a voter committed
// elsewhere is weight the local winner can never gain.
slot_state.observed_votes[vote.voter_id()] = vote;
if ( vote.proposal_id() != slot_state.best_proposal_id )
{
++slot_state.dissent_seen;
}
if ( !slot_state.candidates_frozen || slot_state.best_proposal_id != vote.proposal_id() )
{
logger_->debug( "{}: retained vote for proposal that has not won local slot arbitration "
"proposal_id={} voter_id={}",
__func__,
vote.proposal_id().substr( 0, 8 ),
vote.voter_id().substr( 0, 8 ) );
return;
}
}
if ( proposal_state.seen_voters.find( vote.voter_id() ) != proposal_state.seen_voters.end() )
{
logger_->trace( "{}: ignored: duplicate vote voter_id={}",
__func__,
vote.voter_id().substr( 0, 8 ) );
return;
}
auto proposal_registry_result = LoadRegistryByCidCached( proposal_state.proposal.registry_cid() );
if ( proposal_registry_result.has_error() )
{
logger_->warn( "{}: deferred vote: registry load error={} proposal_id={}",
__func__,
proposal_registry_result.error().message(),
vote.proposal_id().substr( 0, 8 ) );
pending_votes_[vote.proposal_id()].push_back( vote );
return;
}
const auto &proposal_registry = proposal_registry_result.value();
if ( proposal_state.proposal.registry_epoch() != proposal_registry.epoch() )
{
logger_->error( "{}: rejected: registry mismatch proposal_id={}",
__func__,
vote.proposal_id().substr( 0, 8 ) );
return;
}
const auto *validator = registry_->FindValidator( proposal_registry, vote.voter_id() );
const bool is_active_validator = validator && validator->status() == ValidatorRegistry::Status::ACTIVE;
if ( it->second.total_weight == 0 )
{
it->second.total_weight = registry_->TotalWeight( proposal_registry );
}
it->second.votes.push_back( vote );
it->second.seen_voters.insert( vote.voter_id() );
if ( is_active_validator )
{
it->second.approved_weight += validator->weight();
has_quorum = registry_->IsQuorum( it->second.approved_weight, it->second.total_weight );
if ( !has_quorum && IsBridgeMintSubject( proposal_state.proposal ) )
{
// Phase 6 (D-06): bridge-mint subjects decide quorum through the
// cumulative slot tally, not the single-pool weight sum. Recompute
// over the full admitted vote vector so slot dedup/grouping rules
// see every vote (admission here is O(votes^2); acceptable for the
// validator set sizes in play).
const auto slot_result = registry_->EvaluateSlotQuorum( it->second.votes, proposal_registry );
has_quorum = slot_result.has_quorum;
}
if ( has_quorum )
{
if ( !it->second.quorum_reached )
{
it->second.quorum_reached = true;
it->second.quorum_reached_ts_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
std::chrono::system_clock::now().time_since_epoch() )
.count();
}
logger_->debug(
"{}: quorum reached; certificate will be created by timer proposal_id={}",
__func__,
vote.proposal_id() );
}
}
else
{
logger_->debug( "{}: accepted vote from non-validator voter_id={}",
__func__,
vote.voter_id().substr( 0, 8 ) );
}
}
if ( has_quorum )
{
certificates_pending_.store( true );
timer_cv_.notify_all();
}
}
void ConsensusManager::HandleVoteBundle( const VoteBundle &bundle )
{
logger_->trace( "{}: called proposal_id={} votes={}",
__func__,
bundle.proposal_id().substr( 0, 8 ),
bundle.votes_size() );
for ( const auto &vote : bundle.votes() )
{
logger_->trace( "{}: processing voter_id={}", __func__, vote.voter_id().substr( 0, 8 ) );
HandleVote( vote );
}
}
void ConsensusManager::HandleCertificate( const Certificate &certificate )
{
logger_->trace( "{}: called proposal_id={}", __func__, certificate.proposal_id() );
if ( ValidateCertificate( certificate ) != Check::Approve )
{
logger_->error( "{}: rejected: invalid certificate proposal_id={}",
__func__,
certificate.proposal_id() );
return;
}
ProposalState proposal_state;
auto fetch_proposal_state_ret = FetchProposalState( certificate );
if ( fetch_proposal_state_ret.has_value() )
{
proposal_state = fetch_proposal_state_ret.value();
logger_->debug( "{}: fetched proposal state, proposal_id={}",
__func__,
certificate.proposal_id() );
}
else
{
logger_->debug( "{}: proposal state not found, creating new one proposal_id={}",
__func__,
certificate.proposal_id() );
proposal_state = CreateProposalState( certificate );
}
if ( !ValidateCertificateBestProposal( proposal_state, certificate ) )
{
return;
}
ClearProposalSlot( certificate.proposal() );
logger_->debug( "{}: success proposal_id={}", __func__, certificate.proposal_id() );
}
outcome::result<ConsensusManager::ProposalState> ConsensusManager::FetchProposalState(
const Certificate &certificate )
{
std::lock_guard lock( proposals_mutex_ );
auto it = proposals_.find( certificate.proposal_id() );
if ( it == proposals_.end() )
{
return outcome::failure( std::errc::no_such_device );
}
return it->second;
}
ConsensusManager::ProposalState ConsensusManager::CreateProposalState( const Certificate &certificate )
{
ProposalState new_state;
new_state.proposal = certificate.proposal();
new_state.slot_key = GetSlotKey( new_state.proposal );
// Runs on the pubsub receive thread; `proposals_`/`slot_states_` are also mutated
// by the round-timer thread and other pubsub callbacks under `proposals_mutex_`.
{
std::lock_guard lock( proposals_mutex_ );
proposals_.emplace( new_state.proposal.proposal_id(), new_state );
auto &slot_state = slot_states_[new_state.slot_key];
if ( slot_state.best_proposal_id.empty() )
{
slot_state.best_proposal_id = new_state.proposal.proposal_id();
auto nonce_payload = DecodeNonceSubject( new_state.proposal.subject() );
if ( nonce_payload.has_value() )
{
slot_state.best_tx_hash = nonce_payload.value().tx_hash();
}
}
}
return new_state;
}
bool ConsensusManager::ValidateCertificateBestProposal( const ProposalState &state,
const Certificate &certificate ) const
{
if ( certificate.has_proposal() && certificate.proposal().has_subject() &&
DecodeRegistryBatchSubject( certificate.proposal().subject() ).has_value() )
{
// Registry-batch subjects can have multiple competing proposals for the same deterministic batch root.
// Once a valid certificate exists, accept it even if local best_proposal_id changed due proposal races.
return true;
}
std::lock_guard lock( proposals_mutex_ );
auto slot_it = slot_states_.find( state.slot_key );
if ( slot_it != slot_states_.end() && slot_it->second.best_proposal_id != certificate.proposal_id() )
{
logger_->error( "{}: rejected: not best proposal proposal_id={}",
__func__,
certificate.proposal_id() );
return false;
}
return true;
}
std::vector<ConsensusManager::Vote> ConsensusManager::CollectCertificateVotes(
const Certificate &certificate ) const
{
std::vector<Vote> votes;
votes.reserve( static_cast<size_t>( certificate.votes_size() ) );
for ( const auto &vote : certificate.votes() )
{
logger_->trace( "{}: processing vote voter_id={}", __func__, vote.voter_id() );
votes.push_back( vote );
}
return votes;
}
void ConsensusManager::ClearProposalSlot( const Proposal &proposal )
{
std::lock_guard lock( proposals_mutex_ );
std::string slot_key;
auto it = proposals_.find( proposal.proposal_id() );
if ( it != proposals_.end() )
{
slot_key = it->second.slot_key;
}
else
{
slot_key = GetSlotKey( proposal );
}
std::unordered_set<std::string> ids_to_remove;
ids_to_remove.insert( proposal.proposal_id() );
for ( const auto &kv : proposals_ )
{
if ( kv.second.slot_key == slot_key )
{
ids_to_remove.insert( kv.first );
}
}
for ( const auto &proposal_id : ids_to_remove )
{
RemovePendingProposalLocked( proposal_id, "slot-cleanup" );
proposals_.erase( proposal_id );
}
slot_states_.erase( slot_key );
bool has_pending = false;
for ( const auto &kv : proposals_ )
{
if ( kv.second.quorum_reached )
{
has_pending = true;
break;
}
}
certificates_pending_.store( has_pending );
if ( !has_pending )
{
timer_cv_.notify_all();
}
}
std::string ConsensusManager::GetSlotKey( const Proposal &proposal )
{
ConsensusManagerLogger()->trace( "{}: called proposal_id={}", __func__, proposal.proposal_id() );
if ( !proposal.subject().has_subject_type_hash() )
{
return proposal.proposal_id();
}
const auto &subject = proposal.subject();
const auto &hash = subject.subject_type_hash().hash();
{
std::shared_lock lock( slot_key_handlers_mutex_ );
auto it = slot_key_handlers_.find( hash );
if ( it != slot_key_handlers_.end() )
{
return it->second( subject );
}
}
auto subject_id = ComputeSubjectId( subject );
return subject_id.has_value() ? subject_id.value() : proposal.proposal_id();
}
bool ConsensusManager::IsBetterProposal( const Proposal &candidate, const Proposal ¤t ) const
{
logger_->trace( "{}: called candidate={} current={}",
__func__,
candidate.proposal_id(),
current.proposal_id() );
auto candidate_nonce = DecodeNonceSubject( candidate.subject() );
auto current_nonce = DecodeNonceSubject( current.subject() );
if ( candidate_nonce.has_value() && current_nonce.has_value() )
{
const auto &cand_hash = candidate_nonce.value().tx_hash();
const auto &curr_hash = current_nonce.value().tx_hash();
if ( cand_hash == curr_hash )
{
return candidate.proposal_id() < current.proposal_id();
}
return BestHash( curr_hash, cand_hash ) == cand_hash;
}
return candidate.proposal_id() < current.proposal_id();
}
const std::string &ConsensusManager::BestHash( const std::string &a, const std::string &b )
{
return ( a <= b ) ? a : b;
}
outcome::result<std::string> ConsensusManager::ComputeSubjectId( const Subject &subject )
{
ConsensusManagerLogger()->trace( "{}: called", __func__ );
std::string serialized;
if ( !subject.SerializeToString( &serialized ) )
{
ConsensusManagerLogger()->error( "{}: failed: serialization error", __func__ );
return outcome::failure( std::errc::invalid_argument );
}
auto hash = sgns::crypto::sha2_256( serialized.data(), serialized.size() );
ConsensusManagerLogger()->debug( "{}: success", __func__ );
return base::hex_lower( gsl::span<const uint8_t>( hash.data(), hash.size() ) );
}
namespace
{
constexpr size_t kSubjectTypeHashSize = base::Hash256::size();
outcome::result<std::string> ComputePayloadHash( const std::string &payload )
{
if ( payload.empty() )
{
return outcome::failure( std::errc::invalid_argument );
}
auto hash = sgns::crypto::sha2_256( payload.data(), payload.size() );
return std::string( reinterpret_cast<const char *>( hash.data() ), hash.size() );
}
bool SetSubjectPayload( ConsensusSubject *subject,
const std::string &subject_type_hash,
const google::protobuf::MessageLite &payload )
{
if ( subject == nullptr || subject_type_hash.size() != kSubjectTypeHashSize )
{
return false;
}
std::string serialized;
if ( !payload.SerializeToString( &serialized ) )
{
return false;
}
std::string canonical_payload = subject_type_hash + serialized;
auto payload_hash = ComputePayloadHash( canonical_payload );
if ( payload_hash.has_error() )
{
return false;
}
subject->set_payload( canonical_payload.data(), canonical_payload.size() );
subject->set_payload_hash( payload_hash.value().data(), payload_hash.value().size() );
return true;
}
outcome::result<std::string> ExtractBuiltinPayload( const ConsensusSubject &subject,
std::string_view subject_type )
{
auto expected = ConsensusManager::ComputeSubjectTypeHash( subject_type );
if ( expected.has_error() || !subject.has_subject_type_hash() ||
subject.subject_type_hash().hash() != expected.value() ||
subject.payload().size() <= kSubjectTypeHashSize || expected.value().size() != kSubjectTypeHashSize ||
subject.payload().compare( 0, kSubjectTypeHashSize, expected.value() ) != 0 )
{
return outcome::failure( std::errc::invalid_argument );
}
return subject.payload().substr( kSubjectTypeHashSize );
}
}
outcome::result<std::string> ConsensusManager::ComputeSubjectTypeHash( std::string_view subject_type )
{
if ( subject_type.empty() )
{
return outcome::failure( std::errc::invalid_argument );
}
auto hash = sgns::crypto::sha2_256( subject_type.data(), subject_type.size() );
return std::string( reinterpret_cast<const char *>( hash.data() ), hash.size() );
}
bool ConsensusManager::SubjectTypeMatches( const Subject &subject, std::string_view subject_type )
{
auto expected = ComputeSubjectTypeHash( subject_type );
return expected.has_value() && subject.has_subject_type_hash() &&
subject.subject_type_hash().hash() == expected.value();
}
outcome::result<NonceSubject> ConsensusManager::DecodeNonceSubject( const Subject &subject )
{
auto raw_payload = ExtractBuiltinPayload( subject, NONCE_SUBJECT_TYPE );
if ( raw_payload.has_error() )
{
return outcome::failure( raw_payload.error() );
}
NonceSubject payload;
if ( !payload.ParseFromString( raw_payload.value() ) )
{
return outcome::failure( std::errc::invalid_argument );
}
return payload;
}
outcome::result<TaskResultSubject> ConsensusManager::DecodeTaskResultSubject( const Subject &subject )
{
auto raw_payload = ExtractBuiltinPayload( subject, TASK_RESULT_SUBJECT_TYPE );
if ( raw_payload.has_error() )
{
return outcome::failure( raw_payload.error() );
}
TaskResultSubject payload;
if ( !payload.ParseFromString( raw_payload.value() ) )
{
return outcome::failure( std::errc::invalid_argument );
}
return payload;
}
outcome::result<RegistryBatchSubject> ConsensusManager::DecodeRegistryBatchSubject( const Subject &subject )
{
auto raw_payload = ExtractBuiltinPayload( subject, REGISTRY_BATCH_SUBJECT_TYPE );
if ( raw_payload.has_error() )
{
return outcome::failure( raw_payload.error() );
}
RegistryBatchSubject payload;
if ( !payload.ParseFromString( raw_payload.value() ) )
{
return outcome::failure( std::errc::invalid_argument );
}
return payload;
}
bool ConsensusManager::SubjectHasValidTypeHash( Subject *subject )
{
return subject != nullptr && subject->has_subject_type_hash() && !subject->subject_type_hash().hash().empty();
}
outcome::result<ConsensusManager::Subject> ConsensusManager::CreateNonceSubject(
const std::string &account_id,
uint64_t nonce,
const std::string &tx_hash,
const EmbeddedTransaction &transaction,
const std::optional<UTXOTransitionCommitment> &utxo_commitment,
const std::optional<UTXOWitness> &utxo_witness )
{
ConsensusManagerLogger()->trace( "{}: called account_id={} nonce={}", __func__, account_id, nonce );
Subject subject;
subject.set_account_id( account_id );
NonceSubject payload;
payload.set_nonce( nonce );
payload.set_tx_hash( tx_hash.data(), tx_hash.size() );
*payload.mutable_transaction() = transaction;
if ( utxo_commitment.has_value() )
{
*payload.mutable_utxo_commitment() = utxo_commitment.value();
}
if ( utxo_witness.has_value() )
{
*payload.mutable_utxo_witness() = utxo_witness.value();
}
auto type_hash = ComputeSubjectTypeHash( NONCE_SUBJECT_TYPE );
if ( type_hash.has_error() || !SetSubjectPayload( &subject, type_hash.value(), payload ) )
{
return outcome::failure( std::errc::invalid_argument );
}
subject.mutable_subject_type_hash()->set_hash( type_hash.value().data(), type_hash.value().size() );
ConsensusManagerLogger()->debug( "{}: success", __func__ );
return subject;
}
outcome::result<ConsensusManager::Subject> ConsensusManager::CreateTaskResultSubject(
const std::string &account_id,
const std::string &escrow_path,
const std::string &task_result_hash,
uint64_t result_epoch )
{
ConsensusManagerLogger()->trace( "{}: called account_id={} result_epoch={}",
__func__,
account_id,
result_epoch );
Subject subject;
subject.set_account_id( account_id );
TaskResultSubject payload;
payload.set_escrow_path( escrow_path );
payload.set_task_result_hash( task_result_hash.data(), task_result_hash.size() );
payload.set_result_epoch( result_epoch );
auto type_hash = ComputeSubjectTypeHash( TASK_RESULT_SUBJECT_TYPE );
if ( type_hash.has_error() || !SetSubjectPayload( &subject, type_hash.value(), payload ) )
{
return outcome::failure( std::errc::invalid_argument );
}
subject.mutable_subject_type_hash()->set_hash( type_hash.value().data(), type_hash.value().size() );
ConsensusManagerLogger()->debug( "{}: success", __func__ );
return subject;
}
outcome::result<std::string> ConsensusManager::ComputeBatchRoot( const std::vector<std::string> &members )
{
ConsensusManagerLogger()->trace( "{}: called members={}", __func__, members.size() );
if ( members.empty() )
{
ConsensusManagerLogger()->error( "{}: empty member list", __func__ );
return outcome::failure( std::errc::invalid_argument );
}
std::vector<std::string> sorted_members( members.begin(), members.end() );
std::sort( sorted_members.begin(), sorted_members.end() );
std::string payload;
payload += sorted_members[0];
for ( size_t i = 1; i < sorted_members.size(); ++i )
{
payload.push_back( '\n' );
payload += sorted_members[i];
}
auto hash = sgns::crypto::sha2_256( payload.data(), payload.size() );
return base::hex_lower( gsl::span<const uint8_t>( hash.data(), hash.size() ) );
}
outcome::result<ConsensusManager::Subject> ConsensusManager::CreateRegistryBatchSubject(
const std::string &account_id,
const std::string &base_registry_cid,
uint64_t base_registry_epoch,
uint64_t target_registry_epoch,
uint32_t certificate_count,
const std::string &batch_root,
const std::vector<std::string> &member_slots )
{
ConsensusManagerLogger()->trace( "{}: called account_id={} base_epoch={} target_epoch={} certificates={}",
__func__,
account_id.substr( 0, 8 ),
base_registry_epoch,
target_registry_epoch,
certificate_count );
if ( member_slots.empty() || member_slots.size() != static_cast<size_t>( certificate_count ) )
{
ConsensusManagerLogger()->error( "{}: member slot list must match certificate_count={} slots={}",
__func__,
certificate_count,
member_slots.size() );
return outcome::failure( std::errc::invalid_argument );
}
for ( const auto &slot : member_slots )
{
if ( slot.empty() )
{
ConsensusManagerLogger()->error( "{}: member slot list contains an empty slot", __func__ );
return outcome::failure( std::errc::invalid_argument );
}
}
Subject subject;
subject.set_account_id( account_id );
RegistryBatchSubject payload;
payload.set_base_registry_cid( base_registry_cid );
payload.set_base_registry_epoch( base_registry_epoch );
payload.set_target_registry_epoch( target_registry_epoch );
payload.set_certificate_count( certificate_count );
payload.set_batch_root( batch_root.data(), batch_root.size() );
for ( const auto &slot : member_slots )
{
*payload.add_member_certificate_slots() = slot;
}
auto type_hash = ComputeSubjectTypeHash( REGISTRY_BATCH_SUBJECT_TYPE );
if ( type_hash.has_error() || !SetSubjectPayload( &subject, type_hash.value(), payload ) )
{
return outcome::failure( std::errc::invalid_argument );
}
subject.mutable_subject_type_hash()->set_hash( type_hash.value().data(), type_hash.value().size() );
ConsensusManagerLogger()->debug( "{}: success", __func__ );
return subject;
}
outcome::result<ConsensusManager::Subject> ConsensusManager::CreateGenericSubject(
const std::string &account_id,
std::string_view subject_type,
const std::vector<uint8_t> &payload )
{
ConsensusManagerLogger()->trace( "{}: called account_id={} subject_type={}",
__func__,
account_id.substr( 0, 8 ),
subject_type );
if ( account_id.empty() || subject_type.empty() || payload.empty() )
{
return outcome::failure( std::errc::invalid_argument );
}
Subject subject;
subject.set_account_id( account_id );
subject.set_payload( payload.data(), payload.size() );
auto payload_hash = ComputePayloadHash( subject.payload() );
auto type_hash = ComputeSubjectTypeHash( subject_type );
if ( payload_hash.has_error() || type_hash.has_error() )
{
return outcome::failure( std::errc::invalid_argument );
}
subject.set_payload_hash( payload_hash.value().data(), payload_hash.value().size() );
subject.mutable_subject_type_hash()->set_hash( type_hash.value().data(), type_hash.value().size() );
ConsensusManagerLogger()->debug( "{}: success", __func__ );
return subject;
}
std::string ConsensusManager::CreateProposalId( const Proposal &proposal )
{
ConsensusManagerLogger()->trace( "{}: Creating proposal ID", __func__ );
// Proposal ID must be derived from the proposal contents excluding the proposal_id itself.
Proposal copy = proposal;
copy.clear_proposal_id();
auto signing_bytes = ProposalSigningBytes( copy );
if ( signing_bytes.has_error() )
{
ConsensusManagerLogger()->error( "{}: failed, no proposal ID created: signing bytes error={}",
__func__,
signing_bytes.error().message() );
return {};
}
auto hash = sgns::crypto::sha2_256( signing_bytes.value().data(), signing_bytes.value().size() );
auto proposal_id = base::hex_lower( gsl::span<const uint8_t>( hash.data(), hash.size() ) );
ConsensusManagerLogger()->debug( "{}: Proposal ID {} created", __func__, proposal_id.substr( 0, 8 ) );
return proposal_id;
}
bool ConsensusManager::ValidateSubject( const Subject &subject )
{
ConsensusManagerLogger()->trace( "{}: called", __func__ );
if ( subject.account_id().empty() )
{
return false;
}
if ( !subject.has_subject_type_hash() ||
subject.subject_type_hash().hash().size() != base::Hash256::size() )
{
return false;
}
if ( subject.payload().empty() || subject.payload_hash().empty() )
{
return false;
}
auto payload_hash = ComputePayloadHash( subject.payload() );
if ( payload_hash.has_error() || payload_hash.value() != subject.payload_hash() )
{
return false;
}
if ( SubjectTypeMatches( subject, NONCE_SUBJECT_TYPE ) )
{
auto payload = DecodeNonceSubject( subject );
if ( payload.has_error() || payload.value().tx_hash().empty() )
{
return false;
}
if ( payload.value().has_utxo_witness() && !payload.value().has_utxo_commitment() )
{
return false;
}
return true;
}
if ( SubjectTypeMatches( subject, TASK_RESULT_SUBJECT_TYPE ) )
{
auto payload = DecodeTaskResultSubject( subject );
return payload.has_value() && !payload.value().task_result_hash().empty();
}
if ( SubjectTypeMatches( subject, REGISTRY_BATCH_SUBJECT_TYPE ) )
{
auto payload = DecodeRegistryBatchSubject( subject );
if ( payload.has_error() )
{
return false;
}
return !payload.value().base_registry_cid().empty() &&
payload.value().target_registry_epoch() == payload.value().base_registry_epoch() + 1 &&
payload.value().certificate_count() > 0 && !payload.value().batch_root().empty() &&
RegistryBatchSubjectMembersValid( payload.value() );
}
return true;
}
void ConsensusManager::OnConsensusMessage( boost::optional<const ipfs_pubsub::GossipPubSub::Message &> message )
{
logger_->trace( "{}: called", __func__ );
if ( !message )
{
logger_->error( "{}: ignored: message is empty", __func__ );
return;
}
ConsensusMessage decoded;
if ( !decoded.ParseFromArray( message->data.data(), static_cast<int>( message->data.size() ) ) )
{
logger_->error( "{}: Failed to decode consensus message", __func__ );
return;
}
if ( decoded.has_proposal() )
{
logger_->debug( "{}: decoded proposal", __func__ );
HandleProposal( decoded.proposal() );
return;
}
if ( decoded.has_vote() )
{
logger_->debug( "{}: decoded vote", __func__ );
HandleVote( decoded.vote() );
return;
}
// The ConsensusMessage oneof no longer carries a vote_bundle member
// (field 3 reserved on develop); bundles are only assembled locally.
if ( decoded.has_certificate() )
{
logger_->debug( "{}: decoded certificate", __func__ );
HandleCertificate( decoded.certificate() );
}
}
bool ConsensusManager::CheckSubject( const Subject &subject )
{
ConsensusManagerLogger()->trace( "{}: called", __func__ );
if ( subject.account_id().empty() )
{
ConsensusManagerLogger()->error( "{}: subject account_id is empty", __func__ );
return false;
}
if ( !subject.has_subject_type_hash() ||
subject.subject_type_hash().hash().size() != base::Hash256::size() )
{
ConsensusManagerLogger()->error( "{}: subject subject_type_hash is empty or not exactly {} bytes",
__func__,
base::Hash256::size() );
return false;
}
if ( subject.payload().empty() )
{
ConsensusManagerLogger()->error( "{}: subject payload is empty", __func__ );
return false;
}
if ( subject.payload_hash().empty() )
{
ConsensusManagerLogger()->error( "{}: subject payload_hash is empty", __func__ );
return false;
}
auto payload_hash = ComputePayloadHash( subject.payload() );
if ( payload_hash.has_error() || payload_hash.value() != subject.payload_hash() )
{
ConsensusManagerLogger()->error( "{}: subject payload_hash mismatch", __func__ );
return false;
}
if ( SubjectTypeMatches( subject, NONCE_SUBJECT_TYPE ) )
{
auto payload = DecodeNonceSubject( subject );
if ( payload.has_error() || payload.value().tx_hash().empty() )
{
ConsensusManagerLogger()->error( "{}: subject nonce tx_hash is empty", __func__ );
return false;
}
return true;
}
if ( SubjectTypeMatches( subject, TASK_RESULT_SUBJECT_TYPE ) )
{
auto payload = DecodeTaskResultSubject( subject );
if ( payload.has_error() || payload.value().escrow_path().empty() )
{
ConsensusManagerLogger()->error( "{}: subject task_result escrow_path is empty", __func__ );
return false;
}
if ( payload.value().task_result_hash().empty() )
{
ConsensusManagerLogger()->error( "{}: subject task_result task_result_hash is empty", __func__ );
return false;
}
return true;
}
if ( SubjectTypeMatches( subject, REGISTRY_BATCH_SUBJECT_TYPE ) )
{
auto payload = DecodeRegistryBatchSubject( subject );
if ( payload.has_error() )
{
return false;
}
if ( payload.value().base_registry_cid().empty() )
{
ConsensusManagerLogger()->error( "{}: subject registry_batch base_registry_cid is empty", __func__ );
return false;
}
if ( payload.value().target_registry_epoch() != payload.value().base_registry_epoch() + 1 )
{
ConsensusManagerLogger()->error( "{}: subject registry_batch target epoch mismatch", __func__ );
return false;
}
if ( payload.value().certificate_count() == 0 )
{
ConsensusManagerLogger()->error( "{}: subject registry_batch certificate_count is zero", __func__ );
return false;
}
if ( payload.value().batch_root().empty() )
{
ConsensusManagerLogger()->error( "{}: subject registry_batch batch_root is empty", __func__ );
return false;
}
if ( !RegistryBatchSubjectMembersValid( payload.value() ) )
{
ConsensusManagerLogger()->error( "{}: subject registry_batch member slots are invalid", __func__ );
return false;
}
}
return true;
}
bool ConsensusManager::CheckProposal( const Proposal &proposal )
{
if ( proposal.proposal_id().empty() )
{
ConsensusManagerLogger()->error( "{}: Proposal ID missing ", __func__ );
return false;
}
if ( proposal.proposer_id().empty() )
{
ConsensusManagerLogger()->error( "{}: Proposer ID missing ", __func__ );
return false;
}
if ( proposal.registry_cid().empty() )
{
ConsensusManagerLogger()->error( "{}: Registry CID missing ", __func__ );
return false;
}
if ( !proposal.has_subject() )
{
ConsensusManagerLogger()->error( "{}: Proposal without subject ", __func__ );
return false;
}
auto signing_bytes = ProposalSigningBytes( proposal );
if ( signing_bytes.has_error() )
{
ConsensusManagerLogger()->error( "{}: rejected: signing bytes error={}",
__func__,
signing_bytes.error().message() );
return false;
}
if ( !GeniusAccount::VerifySignature( proposal.proposer_id(), proposal.signature(), signing_bytes.value() ) )
{
ConsensusManagerLogger()->error( "{}: rejected: signature verification failed proposer_id={}",
__func__,
proposal.proposer_id() );
return false;
}
return true;
}
bool ConsensusManager::CheckVote( const Vote &vote )
{
if ( vote.proposal_id().empty() )
{
ConsensusManagerLogger()->error( "{}: Vote proposal ID missing ", __func__ );
return false;
}
if ( vote.voter_id().empty() )
{
ConsensusManagerLogger()->error( "{}: Vote voter ID missing ", __func__ );
return false;
}
return true;
}
void ConsensusManager::RecoverPendingCertificateWork()
{
// Timer and post-registration recovery can run concurrently. Keep one
// durable readback-to-handler dispatch in flight so a stalled entry is
// claimed by this manager until it is explicitly stalled again or done.
std::unique_lock recovery_lock( certificate_recovery_mutex_ );
// Compiled once: CRDTWorkJournal takes an optional<std::regex> pattern and
// re-compiling on every timer tick would dominate the recovery path.
static std::regex PATTERN{ CERT_KEY_PATTERN.data(), CERT_KEY_PATTERN.size() };
auto recovered = certificate_work_journal_->RecoverStaleProcessing( PATTERN, std::chrono::seconds( 15 ) );
if ( recovered > 0 )
{
logger_->info( "{}: recovered {} stale certificate work items", __func__, recovered );
}
auto unfinished = certificate_work_journal_->ListUnfinished( PATTERN );
for ( const auto &entry : unfinished )
{
DispatchStalledCertificateEntryLocked( entry );
}
}
void ConsensusManager::DispatchStalledCertificateEntryLocked( const crdt::CRDTWorkJournal::Entry &entry )
{
if ( entry.key.empty() )
{
return;
}
// Seen entries are dispatchable too: the element callback that would
// stall them is skipped whenever the merge is a no-op (e.g. the same
// certificate arriving in a second delta), which would otherwise leave
// committed work wedged in Seen forever.
if ( entry.state == crdt::CRDTWorkJournal::State::Processing )
{
return;
}
const auto now_ms = static_cast<uint64_t>(
std::chrono::duration_cast<std::chrono::milliseconds>( std::chrono::system_clock::now().time_since_epoch() )
.count() );
if ( entry.lease_until_ms != 0 && entry.lease_until_ms > now_ms )
{
return;
}
auto value = db_->Get( { entry.key } );
if ( value.has_error() )
{
certificate_work_journal_->MarkStalled( entry.key, std::chrono::milliseconds( 0 ) );
return;
}
Certificate certificate;
if ( !certificate.ParseFromArray( value.value().data(), value.value().size() ) ||
!ValidateCertificateKey( certificate, entry.key ) )
{
// The durable record is permanently invalid; retire the work instead
// of spinning on it every tick.
certificate_work_journal_->MarkDone( entry.key );
return;
}
const auto validation = ValidateCertificate( certificate );
if ( validation == Check::Stalled )
{
// Registry-dependent quorum still deferred; keep the retry loop alive.
certificate_work_journal_->MarkStalled( entry.key, std::chrono::milliseconds( 0 ) );
return;
}
if ( validation != Check::Approve )
{
certificate_work_journal_->MarkDone( entry.key );
return;
}
ProcessCommittedCertificate( entry.key, certificate );
}
void ConsensusManager::DispatchCertificateWorkForSlot( const std::string &slot_key )
{
if ( slot_key.empty() )
{
return;
}
const auto key = std::string{ CERTIFICATE_BASE_PATH_KEY } + slot_key;
std::unique_lock recovery_lock( certificate_recovery_mutex_ );
auto entry = certificate_work_journal_->GetEntry( key );
if ( !entry.has_value() )
{
return;
}
DispatchStalledCertificateEntryLocked( entry.value() );
}
void ConsensusManager::ProcessCommittedCertificate( const std::string &key, const Certificate &certificate )
{
auto subject_hash = GetSubjectHash( certificate.proposal().subject() );
if ( subject_hash.has_error() )
{
certificate_work_journal_->MarkStalled( key, std::chrono::milliseconds( 0 ) );
return;
}
registry_->OnFinalizedCertificate( certificate );
CertificateSubjectHandler handler;
{
std::shared_lock lock( certificate_handlers_mutex_ );
auto it = certificate_subject_handlers_.find( certificate.proposal().subject().subject_type_hash().hash() );
if ( it == certificate_subject_handlers_.end() )
{
certificate_work_journal_->MarkStalled( key, std::chrono::milliseconds( 0 ) );
logger_->warn( "{}: No subject handler for certificate with key {} ", __func__, key );
return;
}
handler = it->second;
}
const auto slot_key = GetSlotKey( certificate.proposal() );
auto release = ReleaseActiveVoteForAcceptedSlot( slot_key );
if ( release.has_error() )
{
// A read/decode/remove failure leaves the durable certificate work retryable.
certificate_work_journal_->MarkStalled( key, std::chrono::milliseconds( 0 ) );
return;
}
// A successful durable readback proves finality even when this node never
// held a local active-vote record (or already removed it on an earlier replay).
ClearProposalSlot( certificate.proposal() );
{
std::lock_guard lock( fault_test_mutex_ );
++fault_test_counters_.accepted_certificate_readbacks;
}
if ( !EnterFinalityFaultBarrier( accepted_certificate_barrier_ ) )
{
certificate_work_journal_->MarkStalled( key, std::chrono::milliseconds( 0 ) );
return;
}
auto certificate_handler_result = handler( subject_hash.value(), certificate );
if ( certificate_handler_result.has_error() || certificate_handler_result.value() == Check::Stalled )
{
// Transient handler failures retry on the next 500ms tick exactly as
// before; a handler failing REPEATEDLY (kHandlerFailureFastRetries or
// more) backs off exponentially, because re-running full certificate
// validation plus the handler every tick forever is a CPU/log-spam
// loop. The threshold keeps stall-then-recover fault choreography on
// the fast path; attempt_count counts every journal transition
// (receipt alone adds several), so the shift is clamped from below.
auto entry = certificate_work_journal_->GetEntry( key );
const uint64_t attempts = entry.has_value() ? entry->attempt_count : 0;
if ( attempts <= kHandlerFailureFastRetries )
{
certificate_work_journal_->MarkStalled( key, std::chrono::milliseconds( 0 ) );
return;
}
const auto shift = static_cast<unsigned>(
std::min<uint64_t>( attempts - kHandlerFailureFastRetries, 7 ) );
std::chrono::milliseconds backoff( 500ULL << shift );
backoff = std::min( backoff, std::chrono::milliseconds( 60000 ) );
logger_->warn( "{}: handler failed {} times, backing off {}ms for key {}",
__func__,
attempts,
backoff.count(),
key );
certificate_work_journal_->MarkStalled( key, backoff );
return;
}
(void) certificate_work_journal_->MarkDone( key );
(void) WakePendingDependency( PendingDependencyKey::Certificate( subject_hash.value() ) );
}
outcome::result<ConsensusManager::Certificate> ConsensusManager::GetCertificateBySlot( const std::string &slot_key ) const
{
if ( slot_key.empty() || !db_ )
{
return outcome::failure( std::errc::invalid_argument );
}
const auto key = std::string{ CERTIFICATE_BASE_PATH_KEY } + slot_key;
BOOST_OUTCOME_TRY( auto certificate_data, db_->Get( { key } ) );
Certificate certificate;
if ( !certificate.ParseFromArray( certificate_data.data(), certificate_data.size() ) )
{
logger_->error( "{}: invalid certificate payload key={}", __func__, key );
return outcome::failure( std::errc::invalid_argument );
}
if ( GetExpectedCertificateSlotKey( certificate ) != key )
{
logger_->error( "{}: certificate slot key mismatch expected={}",
__func__,
key );
return outcome::failure( std::errc::invalid_argument );
}
if ( ValidateCertificate( certificate ) != Check::Approve )
{
return outcome::failure( std::errc::invalid_argument );
}
return certificate;
}
bool ConsensusManager::CheckCertificateForSlot( const std::string &slot_key ) const
{
return GetCertificateBySlot( slot_key ).has_value();
}
std::string ConsensusManager::GetPrintableSubjectHash( const Subject &subject )
{
auto subject_hash = GetSubjectHash( subject );
const std::string short_hash = subject_hash.has_value() ? subject_hash.value().substr( 0, 8 ) : "Invalid";
return short_hash;
}
}
Updated on 2026-09-25 at 15:46:11 +0000