ARCHIVE/rippled
1
0
Fork 0
mirror of https://github.com/XRPLF/rippled.git synced 2026-07-29 18:10:34 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
fbd512e51d191cebf700fc31fee6a09aac3fd6dd
rippled/include/xrpl/consensus/Consensus.h

2143 lines
74 KiB
C++
Raw Blame History

#pragma once
#include <xrpld/consensus/ConsensusParms.h>
#include <xrpld/consensus/ConsensusProposal.h>
#include <xrpld/consensus/ConsensusSpanNames.h>
#include <xrpld/consensus/ConsensusTypes.h>
#include <xrpl/basics/Log.h>
#include <xrpl/basics/UnorderedContainers.h>
#include <xrpl/basics/chrono.h>
#include <xrpl/beast/clock/abstract_clock.h>
#include <xrpl/beast/utility/Journal.h>
#include <xrpl/beast/utility/instrumentation.h>
#include <xrpl/consensus/ConsensusParms.h>
#include <xrpl/consensus/ConsensusProposal.h>
#include <xrpl/consensus/ConsensusTypes.h>
#include <xrpl/json/json_value.h>
#include <xrpl/json/json_writer.h>
#include <xrpl/ledger/LedgerTiming.h>
#include <xrpl/telemetry/SpanGuard.h>
#include <algorithm>
#include <chrono>
#include <cstddef>
#include <cstdint>
#include <deque>
#include <map>
#include <memory>
#include <optional>
#include <sstream>
#include <string>
#include <string_view>
#include <utility>
namespace xrpl {
/**
* Determines whether the current ledger should close at this time.
*
* This function should be called when a ledger is open and there is no close
* in progress, or when a transaction is received and no close is in progress.
*
* @param anyTransactions indicates whether any transactions have been received
* @param prevProposers proposers in the last closing
* @param proposersClosed proposers who have currently closed this ledger
* @param proposersValidated proposers who have validated the last closed
* ledger
* @param prevRoundTime time for the previous ledger to reach consensus
* @param timeSincePrevClose time since the previous ledger's (possibly
* rounded) close time
* @param openTime duration this ledger has been open
* @param idleInterval the network's desired idle interval
* @param parms Consensus constant parameters
* @param j journal for logging
* @param clog log object to which to append
*/
bool
shouldCloseLedger(
bool anyTransactions,
std::size_t prevProposers,
std::size_t proposersClosed,
std::size_t proposersValidated,
std::chrono::milliseconds prevRoundTime,
std::chrono::milliseconds timeSincePrevClose,
std::chrono::milliseconds openTime,
std::chrono::milliseconds idleInterval,
ConsensusParms const& parms,
beast::Journal j,
std::unique_ptr<std::stringstream> const& clog = {});
/**
* Determine whether the network reached consensus and whether we joined.
*
* @param prevProposers proposers in the last closing (not including us)
* @param currentProposers proposers in this closing so far (not including us)
* @param currentAgree proposers who agree with us
* @param currentFinished proposers who have validated a ledger after this one
* @param previousAgreeTime how long, in milliseconds, it took to agree on the
* last ledger
* @param currentAgreeTime how long, in milliseconds, we've been trying to
* agree
* @param stalled the network appears to be stalled, where
* neither we nor our peers have changed their vote on any disputes in a
* while. This is undesirable, and should be rare, and will cause us to
* end consensus without 80% agreement.
* @param parms Consensus constant parameters
* @param proposing whether we should count ourselves
* @param j journal for logging
* @param clog log object to which to append
*/
ConsensusState
checkConsensus(
std::size_t prevProposers,
std::size_t currentProposers,
std::size_t currentAgree,
std::size_t currentFinished,
std::chrono::milliseconds previousAgreeTime,
std::chrono::milliseconds currentAgreeTime,
bool stalled,
ConsensusParms const& parms,
bool proposing,
beast::Journal j,
std::unique_ptr<std::stringstream> const& clog = {});
/**
* Generic implementation of consensus algorithm.
*
* Achieves consensus on the next ledger.
*
* Two things need consensus:
*
* 1. The set of transactions included in the ledger.
* 2. The close time for the ledger.
*
* The basic flow:
*
* 1. A call to `startRound` places the node in the `Open` phase. In this
* phase, the node is waiting for transactions to include in its open
* ledger.
* 2. Successive calls to `timerEntry` check if the node can close the ledger.
* Once the node `Close`s the open ledger, it transitions to the
* `Establish` phase. In this phase, the node shares/receives peer
* proposals on which transactions should be accepted in the closed ledger.
* 3. During a subsequent call to `timerEntry`, the node determines it has
* reached consensus with its peers on which transactions to include. It
* transitions to the `Accept` phase. In this phase, the node works on
* applying the transactions to the prior ledger to generate a new closed
* ledger. Once the new ledger is completed, the node shares the validated
* ledger with the network, does some book-keeping, then makes a call to
* `startRound` to start the cycle again.
*
* This class uses a generic interface to allow adapting Consensus for specific
* applications. The Adaptor template implements a set of helper functions that
* plug the consensus algorithm into a specific application. It also identifies
* the types that play important roles in Consensus (transactions, ledgers, ...).
* The code stubs below outline the interface and type requirements. The traits
* types must be copy constructible and assignable.
*
* @warning The generic implementation is not thread safe and the public methods
* are not intended to be run concurrently. When in a concurrent environment,
* the application is responsible for ensuring thread-safety. Simply locking
* whenever touching the Consensus instance is one option.
*
* @code
* // A single transaction
* struct Tx
* {
* // Unique identifier of transaction
* using ID = ...;
*
* ID id() const;
*
* };
*
* // A set of transactions
* struct TxSet
* {
* // Unique ID of TxSet (not of Tx)
* using ID = ...;
* // Type of individual transaction comprising the TxSet
* using Tx = Tx;
*
* bool exists(Tx::ID const &) const;
* // Return value should have semantics like Tx const *
* Tx const * find(Tx::ID const &) const ;
* ID const & id() const;
*
* // Return set of transactions that are not common to this set or other
* // boolean indicates which set it was in
* std::map<Tx::ID, bool> compare(TxSet const & other) const;
*
* // A mutable view of transactions
* struct MutableTxSet
* {
* MutableTxSet(TxSet const &);
* bool insert(Tx const &);
* bool erase(Tx::ID const &);
* };
*
* // Construct from a mutable view.
* TxSet(MutableTxSet const &);
*
* // Alternatively, if the TxSet is itself mutable
* // just alias MutableTxSet = TxSet
*
* };
*
* // Agreed upon state that consensus transactions will modify
* struct Ledger
* {
* using ID = ...;
* using Seq = ...;
*
* // Unique identifier of ledger
* ID const id() const;
* Seq seq() const;
* auto closeTimeResolution() const;
* auto closeAgree() const;
* auto closeTime() const;
* auto parentCloseTime() const;
* json::Value getJson() const;
* };
*
* // Wraps a peer's ConsensusProposal
* struct PeerPosition
* {
* ConsensusProposal<
* std::uint32_t, //NodeID,
* typename Ledger::ID,
* typename TxSet::ID> const &
* proposal() const;
*
* };
*
*
* class Adaptor
* {
* public:
* //-----------------------------------------------------------------------
* // Define consensus types
* using Ledger_t = Ledger;
* using NodeID_t = std::uint32_t;
* using TxSet_t = TxSet;
* using PeerPosition_t = PeerPosition;
*
* //-----------------------------------------------------------------------
* //
* // Attempt to acquire a specific ledger.
* std::optional<Ledger> acquireLedger(Ledger::ID const & ledgerID);
*
* // Acquire the transaction set associated with a proposed position.
* std::optional<TxSet> acquireTxSet(TxSet::ID const & setID);
*
* // Whether any transactions are in the open ledger
* bool hasOpenTransactions() const;
*
* // Number of proposers that have validated the given ledger
* std::size_t proposersValidated(Ledger::ID const & prevLedger) const;
*
* // Number of proposers that have validated a ledger descended from the
* // given ledger; if prevLedger.id() != prevLedgerID, use prevLedgerID
* // for the determination
* std::size_t proposersFinished(Ledger const & prevLedger,
* Ledger::ID const & prevLedger) const;
*
* // Return the ID of the last closed (and validated) ledger that the
* // application thinks consensus should use as the prior ledger.
* Ledger::ID getPrevLedger(Ledger::ID const & prevLedgerID,
* Ledger const & prevLedger,
* Mode mode);
*
* // Called whenever consensus operating mode changes
* void onModeChange(ConsensusMode before, ConsensusMode after);
*
* // Called when ledger closes
* Result onClose(Ledger const &, Ledger const & prev, Mode mode);
*
* // Called when ledger is accepted by consensus
* void onAccept(Result const & result,
* RCLCxLedger const & prevLedger,
* NetClock::duration closeResolution,
* CloseTimes const & rawCloseTimes,
* Mode const & mode);
*
* // Called when ledger was forcibly accepted by consensus via the simulate
* // function.
* void onForceAccept(Result const & result,
* RCLCxLedger const & prevLedger,
* NetClock::duration closeResolution,
* CloseTimes const & rawCloseTimes,
* Mode const & mode);
*
* // Propose the position to peers.
* void propose(ConsensusProposal<...> const & pos);
*
* // Share a received peer proposal with other peer's.
* void share(PeerPosition_t const & prop);
*
* // Share a disputed transaction with peers
* void share(Txn const & tx);
*
* // Share given transaction set with peers
* void share(TxSet const &s);
*
* // Consensus timing parameters and constants
* ConsensusParms const &
* parms() const;
* };
* @endcode
*
* @tparam Adaptor Defines types and provides helper functions needed to adapt
* Consensus to the larger application.
*/
template <class Adaptor>
class Consensus
{
using Ledger_t = Adaptor::Ledger_t;
using TxSet_t = Adaptor::TxSet_t;
using NodeID_t = Adaptor::NodeID_t;
using Tx_t = TxSet_t::Tx;
using PeerPosition_t = Adaptor::PeerPosition_t;
using Proposal_t = ConsensusProposal<NodeID_t, typename Ledger_t::ID, typename TxSet_t::ID>;
using Result = ConsensusResult<Adaptor>;
// Helper class to ensure adaptor is notified whenever the ConsensusMode
// changes
class MonitoredMode
{
ConsensusMode mode_;
public:
MonitoredMode(ConsensusMode m) : mode_{m}
{
}
[[nodiscard]] ConsensusMode
get() const
{
return mode_;
}
void
set(ConsensusMode mode, Adaptor& a)
{
a.onModeChange(mode_, mode);
mode_ = mode;
}
};
public:
/**
* Clock type for measuring time within the consensus code
*/
using clock_type = beast::AbstractClock<std::chrono::steady_clock>;
Consensus(Consensus&&) noexcept = default;
/**
* Constructor.
*
* @param clock The clock used to internally sample consensus progress
* @param adaptor The instance of the adaptor class
* @param j The journal to log debug output
*/
Consensus(clock_type const& clock, Adaptor& adaptor, beast::Journal j);
/**
* Kick-off the next round of consensus.
*
* Called by the client code to start each round of consensus.
*
* @param now The network adjusted time
* @param prevLedgerID the ID of the last ledger
* @param prevLedger The last ledger
* @param nowUntrusted ID of nodes that are newly untrusted this round
* @param proposing Whether we want to send proposals to peers this
* round.
* @param clog log object to which to append
*
* @note @b prevLedgerID is not required to the ID of @b prevLedger since
* the ID may be known locally before the contents of the ledger arrive
*/
void
startRound(
NetClock::time_point const& now,
Ledger_t::ID const& prevLedgerID,
Ledger_t prevLedger,
hash_set<NodeID_t> const& nowUntrusted,
bool proposing,
std::unique_ptr<std::stringstream> const& clog = {});
/**
* A peer has proposed a new position, adjust our tracking.
*
* @param now The network adjusted time
* @param newProposal The new proposal from a peer
* @return Whether we should do delayed relay of this proposal.
*/
bool
peerProposal(NetClock::time_point const& now, PeerPosition_t const& newProposal);
/**
* Call periodically to drive consensus forward.
*
* @param now The network adjusted time
* @param clog log object to which to append
*/
void
timerEntry(
NetClock::time_point const& now,
std::unique_ptr<std::stringstream> const& clog = {});
/**
* Process a transaction set acquired from the network
*
* @param now The network adjusted time
* @param txSet the transaction set
*/
void
gotTxSet(NetClock::time_point const& now, TxSet_t const& txSet);
/**
* Simulate the consensus process without any network traffic.
*
* The end result, is that consensus begins and completes as if everyone
* had agreed with whatever we propose.
*
* This function is only called from the rpc "ledger_accept" path with the
* server in standalone mode and SHOULD NOT be used during the normal
* consensus process.
*
* Simulate will call onForceAccept since clients are manually driving
* consensus to the accept phase.
*
* @param now The current network adjusted time.
* @param consensusDelay Duration to delay between closing and accepting the
* ledger. Uses 100ms if unspecified.
*/
void
simulate(
NetClock::time_point const& now,
std::optional<std::chrono::milliseconds> consensusDelay);
/**
* Get the previous ledger ID.
*
* The previous ledger is the last ledger seen by the consensus code and
* should correspond to the most recent validated ledger seen by this peer.
*
* @return ID of previous ledger
*/
Ledger_t::ID
prevLedgerID() const
{
return prevLedgerID_;
}
[[nodiscard]] ConsensusPhase
phase() const
{
return phase_;
}
/**
* Get the Json state of the consensus process.
*
* Called by the consensus_info RPC.
*
* @param full True if verbose response desired.
* @return The Json state.
*/
[[nodiscard]] json::Value
getJson(bool full) const;
private:
/**
* Why startRoundInternal is being entered.
*
* Distinguishes the normal Initial entry (from public startRound)
* from a Recovered re-entry (from handleWrongLedger after the
* correct prior ledger was acquired mid-round). The Recovered path
* resets phase to Open within the SAME round, which is a state
* transition that would otherwise be invisible in traces — the
* recovery flag drives a `phase.recovery` event on the round span.
*/
enum class StartRoundReason : std::uint8_t {
Initial,
Recovered,
};
void
startRoundInternal(
NetClock::time_point const& now,
Ledger_t::ID const& prevLedgerID,
Ledger_t const& prevLedger,
ConsensusMode mode,
std::unique_ptr<std::stringstream> const& clog,
StartRoundReason reason = StartRoundReason::Initial);
// Change our view of the previous ledger
void
handleWrongLedger(Ledger_t::ID const& lgrId, std::unique_ptr<std::stringstream> const& clog);
/**
* Check if our previous ledger matches the network's.
*
* If the previous ledger differs, we are no longer in sync with
* the network and need to bow out/switch modes.
*/
void
checkLedger(std::unique_ptr<std::stringstream> const& clog);
/**
* If we radically changed our consensus context for some reason,
* we need to replay recent proposals so that they're not lost.
*/
void
playbackProposals();
/**
* Handle a replayed or a new peer proposal.
*/
bool
peerProposalInternal(NetClock::time_point const& now, PeerPosition_t const& newProposal);
/**
* Handle pre-close phase.
*
* In the pre-close phase, the ledger is open as we wait for new
* transactions. After enough time has elapsed, we will close the ledger,
* switch to the establish phase and start the consensus process.
*/
void
phaseOpen(std::unique_ptr<std::stringstream> const& clog);
/**
* Handle establish phase.
*
* In the establish phase, the ledger has closed and we work with peers
* to reach consensus. Update our position only on the timer, and in this
* phase.
*
* If we have consensus, move to the accepted phase.
*/
void
phaseEstablish(std::unique_ptr<std::stringstream> const& clog);
/**
* Evaluate whether pausing increases likelihood of validation.
*
* As a validator that has previously synced to the network, if our most
* recent locally-validated ledger did not also achieve
* full validation, then consider pausing for awhile based on
* the state of other validators.
*
* Pausing may be beneficial in this situation if at least one validator
* is known to be on a sequence number earlier than ours. The minimum
* number of validators on the same sequence number does not guarantee
* consensus, and waiting for all validators may be too time-consuming.
* Therefore, a variable threshold is enacted based on the number
* of ledgers past network validation that we are on. For the first phase,
* the threshold is also the minimum required for quorum. For the last,
* no online validators can have a lower sequence number. For intermediate
* phases, the threshold is linear between the minimum required for
* quorum and 100%. For example, with 3 total phases and a quorum of
* 80%, the 2nd phase would be 90%. Once the final phase is reached,
* if consensus still fails to occur, the cycle is begun again at phase 1.
*
* @return Whether to pause to wait for lagging proposers.
*/
[[nodiscard]] bool
shouldPause(std::unique_ptr<std::stringstream> const& clog) const;
// Close the open ledger and establish initial position.
void
closeLedger(std::unique_ptr<std::stringstream> const& clog);
// Adjust our positions to try to agree with other validators.
void
updateOurPositions(std::unique_ptr<std::stringstream> const& clog);
bool
haveConsensus(std::unique_ptr<std::stringstream> const& clog);
// Create disputes between our position and the provided one.
void
createDisputes(TxSet_t const& o, std::unique_ptr<std::stringstream> const& clog = {});
// Update our disputes given that this node has adopted a new position.
// Will call createDisputes as needed.
void
updateDisputes(NodeID_t const& node, TxSet_t const& other);
// Revoke our outstanding proposal, if any, and cease proposing
// until this round ends.
void
leaveConsensus(std::unique_ptr<std::stringstream> const& clog);
// The rounded or effective close time estimate from a proposer
[[nodiscard]] NetClock::time_point
asCloseTime(NetClock::time_point raw) const;
private:
Adaptor& adaptor_;
ConsensusPhase phase_{ConsensusPhase::Accepted};
MonitoredMode mode_{ConsensusMode::Observing};
bool firstRound_ = true;
bool haveCloseTimeConsensus_ = false;
clock_type const& clock_;
// How long the consensus convergence has taken, expressed as
// a percentage of the time that we expected it to take.
int convergePercent_{0};
// How long has this round been open
ConsensusTimer openTime_;
NetClock::duration closeResolution_ = kLedgerDefaultTimeResolution;
ConsensusParms::AvalancheState closeTimeAvalancheState_ = ConsensusParms::AvalancheState::Init;
// Time it took for the last consensus round to converge
std::chrono::milliseconds prevRoundTime_{};
//-------------------------------------------------------------------------
// Network time measurements of consensus progress
// The current network adjusted time. This is the network time the
// ledger would close if it closed now
NetClock::time_point now_;
NetClock::time_point prevCloseTime_;
//-------------------------------------------------------------------------
// Non-peer (self) consensus data
// Last validated ledger ID provided to consensus
Ledger_t::ID prevLedgerID_;
// Last validated ledger seen by consensus
Ledger_t previousLedger_;
// Transaction Sets, indexed by hash of transaction tree
hash_map<typename TxSet_t::ID, TxSet_t const> acquired_;
std::optional<Result> result_;
ConsensusCloseTimes rawCloseTimes_;
// The number of calls to phaseEstablish where none of our peers
// have changed any votes on disputed transactions.
std::size_t peerUnchangedCounter_ = 0;
// The total number of times we have called phaseEstablish
std::size_t establishCounter_ = 0;
//-------------------------------------------------------------------------
// Peer related consensus data
// Peer proposed positions for the current round
hash_map<NodeID_t, PeerPosition_t> currPeerPositions_;
// Recently received peer positions, available when transitioning between
// ledgers or rounds
hash_map<NodeID_t, std::deque<PeerPosition_t>> recentPeerPositions_;
// The number of proposers who participated in the last consensus round
std::size_t prevProposers_ = 0;
// nodes that have bowed out of this consensus process
hash_set<NodeID_t> deadNodes_;
/**
* Span for the establish phase of consensus.
* Created when the ledger closes and we enter phaseEstablish;
* cleared (ended) when consensus is reached. A thread-free SpanGuard,
* emplaced and reset() on different job workers.
*/
std::optional<xrpl::telemetry::SpanGuard> establishSpan_;
/**
* Captured context of establishSpan_ (its own span context). Children
* (update_positions, check) build from this explicit context.
*/
xrpl::telemetry::SpanContext establishSpanContext_;
/**
* Span for the open phase of consensus.
* Created in startRoundInternal(); cleared (ended) in closeLedger().
* A thread-free SpanGuard, emplaced and reset() on different job workers.
*/
std::optional<xrpl::telemetry::SpanGuard> openSpan_;
/**
* Create the establish-phase span if not yet active.
* Called on each phaseEstablish() invocation; no-op while span is live.
*/
void
startEstablishTracing();
/**
* Overwrite convergence metrics on the establish span each iteration.
* Final span attributes always reflect the last state before consensus.
*/
void
updateEstablishTracing();
/**
* End the establish span when transitioning to the accepted phase.
*/
void
endEstablishTracing();
// Journal for debugging
beast::Journal const j_;
};
template <class Adaptor>
Consensus<Adaptor>::Consensus(clock_type const& clock, Adaptor& adaptor, beast::Journal journal)
: adaptor_(adaptor), clock_(clock), j_{journal}
{
JLOG(j_.debug()) << "Creating consensus object";
}
template <class Adaptor>
void
Consensus<Adaptor>::startRound(
NetClock::time_point const& now,
Ledger_t::ID const& prevLedgerID,
Ledger_t prevLedger,
hash_set<NodeID_t> const& nowUntrusted,
bool proposing,
std::unique_ptr<std::stringstream> const& clog)
{
if (firstRound_)
{
// take our initial view of closeTime_ from the seed ledger
prevRoundTime_ = adaptor_.parms().ledgerIdleInterval;
prevCloseTime_ = prevLedger.closeTime();
firstRound_ = false;
}
else
{
prevCloseTime_ = rawCloseTimes_.self;
}
for (NodeID_t const& n : nowUntrusted)
recentPeerPositions_.erase(n);
ConsensusMode startMode = proposing ? ConsensusMode::Proposing : ConsensusMode::Observing;
// We were handed the wrong ledger
if (prevLedger.id() != prevLedgerID)
{
// try to acquire the correct one
if (auto newLedger = adaptor_.acquireLedger(prevLedgerID))
{
prevLedger = *newLedger;
}
else // Unable to acquire the correct ledger
{
startMode = ConsensusMode::WrongLedger;
JLOG(j_.info()) << "Entering consensus with: " << previousLedger_.id();
JLOG(j_.info()) << "Correct LCL is: " << prevLedgerID;
}
}
startRoundInternal(now, prevLedgerID, prevLedger, startMode, clog);
}
template <class Adaptor>
void
Consensus<Adaptor>::startRoundInternal(
NetClock::time_point const& now,
Ledger_t::ID const& prevLedgerID,
Ledger_t const& prevLedger,
ConsensusMode mode,
std::unique_ptr<std::stringstream> const& clog,
StartRoundReason const reason)
{
// Recovery path: handleWrongLedger acquired the correct prior ledger
// and re-entered startRoundInternal mid-round. The roundSpan_ owned by
// the adaptor is still the SAME span as before — startRoundTracing is
// not called on the recovery path, so we record the recovery as an
// event on the surviving round span. Pass empty phaseLabel to leave
// consensus_phase unchanged (the actual phase reset to open is marked
// separately by the new openSpan_ being emplaced below).
if (reason == StartRoundReason::Recovered)
{
adaptor_.onPhaseEvent(telemetry::consensus::span::event::phaseRecovery, "");
}
phase_ = ConsensusPhase::Open;
JLOG(j_.debug()) << "transitioned to ConsensusPhase::Open ";
CLOG(clog) << "startRoundInternal transitioned to ConsensusPhase::Open, "
"previous ledgerID: "
<< prevLedgerID << ", seq: " << prevLedger.seq() << ". ";
// Reset establishSpan_ so a wrongLedger recovery mid-establish doesn't
// leak the prior round's span into the new one (startEstablishTracing
// early-returns when establishSpan_ is populated).
establishSpan_.reset();
establishSpanContext_ = telemetry::SpanContext{};
// Child of the round span via its captured context: parent phase.open
// explicitly under roundSpanContext_. An invalid round context (round span
// not yet created) yields a null guard. openSpan_ is a thread-free
// SpanGuard emplaced here on one job worker and reset() on another; no
// scope to strip.
openSpan_.emplace(
telemetry::SpanGuard::childSpan(
telemetry::consensus::span::phaseOpen, adaptor_.roundSpanContext()));
// On the Recovered path, fire phase.open here because startRoundTracing
// (which fires it for the Initial path) is not called on re-entry. On
// the Initial path this is a no-op because the round span hasn't been
// created yet — the phase.open event is fired later by startRoundTracing
// after the new round span is in place.
if (reason == StartRoundReason::Recovered)
{
adaptor_.onPhaseEvent(
telemetry::consensus::span::event::phaseOpen,
telemetry::consensus::span::val::phaseOpen);
}
mode_.set(mode, adaptor_);
now_ = now;
prevLedgerID_ = prevLedgerID;
previousLedger_ = prevLedger;
result_.reset();
convergePercent_ = 0;
closeTimeAvalancheState_ = ConsensusParms::AvalancheState::Init;
haveCloseTimeConsensus_ = false;
openTime_.reset(clock_.now());
currPeerPositions_.clear();
acquired_.clear();
rawCloseTimes_.peers.clear();
rawCloseTimes_.self = {};
deadNodes_.clear();
closeResolution_ = getNextLedgerTimeResolution(
previousLedger_.closeTimeResolution(),
previousLedger_.closeAgree(),
previousLedger_.seq() + typename Ledger_t::Seq{1});
playbackProposals();
CLOG(clog) << "number of peer proposals,previous proposers: " << currPeerPositions_.size()
<< ',' << prevProposers_ << ". ";
if (currPeerPositions_.size() > (prevProposers_ / 2))
{
// We may be falling behind, don't wait for the timer
// consider closing the ledger immediately
CLOG(clog) << "consider closing the ledger immediately. ";
timerEntry(now_, clog);
}
}
template <class Adaptor>
bool
Consensus<Adaptor>::peerProposal(NetClock::time_point const& now, PeerPosition_t const& newPeerPos)
{
JLOG(j_.debug()) << "PROPOSAL " << newPeerPos.render();
auto const& peerID = newPeerPos.proposal().nodeID();
// Always need to store recent positions
{
auto& props = recentPeerPositions_[peerID];
if (props.size() >= 10)
props.pop_front();
props.push_back(newPeerPos);
}
return peerProposalInternal(now, newPeerPos);
}
template <class Adaptor>
bool
Consensus<Adaptor>::peerProposalInternal(
NetClock::time_point const& now,
PeerPosition_t const& newPeerPos)
{
// Nothing to do for now if we are currently working on a ledger
if (phase_ == ConsensusPhase::Accepted)
return false;
now_ = now;
auto const& newPeerProp = newPeerPos.proposal();
if (newPeerProp.prevLedger() != prevLedgerID_)
{
JLOG(j_.debug()) << "Got proposal for " << newPeerProp.prevLedger() << " but we are on "
<< prevLedgerID_;
return false;
}
auto const& peerID = newPeerProp.nodeID();
if (deadNodes_.find(peerID) != deadNodes_.end())
{
JLOG(j_.info()) << "Position from dead node: " << peerID;
return false;
}
{
// update current position
auto peerPosIt = currPeerPositions_.find(peerID);
if (peerPosIt != currPeerPositions_.end())
{
if (newPeerProp.proposeSeq() <= peerPosIt->second.proposal().proposeSeq())
{
return false;
}
}
if (newPeerProp.isBowOut())
{
JLOG(j_.info()) << "Peer " << peerID << " bows out";
if (result_)
{
for (auto& it : result_->disputes)
it.second.unVote(peerID);
}
if (peerPosIt != currPeerPositions_.end())
currPeerPositions_.erase(peerID);
deadNodes_.insert(peerID);
return true;
}
if (peerPosIt != currPeerPositions_.end())
{
peerPosIt->second = newPeerPos;
}
else
{
currPeerPositions_.emplace(peerID, newPeerPos);
}
}
if (newPeerProp.isInitial())
{
// Record the close time estimate
JLOG(j_.trace()) << "Peer reports close time as "
<< newPeerProp.closeTime().time_since_epoch().count();
++rawCloseTimes_.peers[newPeerProp.closeTime()];
}
JLOG(j_.trace()) << "Processing peer proposal " << newPeerProp.proposeSeq() << "/"
<< newPeerProp.position();
{
auto const ait = acquired_.find(newPeerProp.position());
if (ait == acquired_.end())
{
// acquireTxSet will return the set if it is available, or
// spawn a request for it and return nullopt/nullptr. It will call
// gotTxSet once it arrives
if (auto set = adaptor_.acquireTxSet(newPeerProp.position()))
{
gotTxSet(now_, *set);
}
else
{
JLOG(j_.debug()) << "Don't have tx set for peer";
}
}
else if (result_)
{
updateDisputes(newPeerProp.nodeID(), ait->second);
}
}
return true;
}
template <class Adaptor>
void
Consensus<Adaptor>::timerEntry(
NetClock::time_point const& now,
std::unique_ptr<std::stringstream> const& clog)
{
CLOG(clog) << "Consensus<Adaptor>::timerEntry. ";
// Nothing to do if we are currently working on a ledger
if (phase_ == ConsensusPhase::Accepted)
{
CLOG(clog) << "Nothing to do during accepted phase. ";
return;
}
now_ = now;
CLOG(clog) << "Set network adjusted time to " << to_string(now) << ". ";
// Check we are on the proper ledger (this may change phase_)
auto const phaseOrig = phase_;
CLOG(clog) << "Phase " << to_string(phaseOrig) << ". ";
checkLedger(clog);
if (phaseOrig != phase_)
{
CLOG(clog) << "Changed phase to << " << to_string(phase_) << ". ";
}
if (phase_ == ConsensusPhase::Open)
{
phaseOpen(clog);
}
else if (phase_ == ConsensusPhase::Establish)
{
phaseEstablish(clog);
}
CLOG(clog) << "timerEntry finishing in phase " << to_string(phase_) << ". ";
}
template <class Adaptor>
void
Consensus<Adaptor>::gotTxSet(NetClock::time_point const& now, TxSet_t const& txSet)
{
// Nothing to do if we've finished work on a ledger
if (phase_ == ConsensusPhase::Accepted)
return;
now_ = now;
auto id = txSet.id();
// If we've already processed this transaction set since requesting
// it from the network, there is nothing to do now
if (!acquired_.emplace(id, txSet).second)
return;
if (!result_)
{
JLOG(j_.debug()) << "Not creating disputes: no position yet.";
}
else
{
// Our position is added to acquired_ as soon as we create it,
// so this txSet must differ
XRPL_ASSERT(
id != result_->position.position(),
"xrpl::Consensus::gotTxSet : updated transaction set");
bool any = false;
for (auto const& [nodeId, peerPos] : currPeerPositions_)
{
if (peerPos.proposal().position() == id)
{
updateDisputes(nodeId, txSet);
any = true;
}
}
if (!any)
{
JLOG(j_.warn()) << "By the time we got " << id << " no peers were proposing it";
}
}
}
template <class Adaptor>
void
Consensus<Adaptor>::simulate(
NetClock::time_point const& now,
std::optional<std::chrono::milliseconds> consensusDelay)
{
using namespace std::chrono_literals;
JLOG(j_.info()) << "Simulating consensus";
now_ = now;
closeLedger({});
// NOLINTBEGIN(bugprone-unchecked-optional-access) closeLedger sets result_
result_->roundTime.tick(consensusDelay.value_or(100ms));
result_->proposers = prevProposers_ = currPeerPositions_.size();
prevRoundTime_ = result_->roundTime.read();
phase_ = ConsensusPhase::Accepted;
adaptor_.onPhaseEvent(
telemetry::consensus::span::event::phaseAccepted,
telemetry::consensus::span::val::phaseAccepted);
adaptor_.onForceAccept(
*result_, previousLedger_, closeResolution_, rawCloseTimes_, mode_.get(), getJson(true));
// NOLINTEND(bugprone-unchecked-optional-access)
JLOG(j_.info()) << "Simulation complete";
}
template <class Adaptor>
json::Value
Consensus<Adaptor>::getJson(bool full) const
{
using std::to_string;
using Int = json::Value::Int;
json::Value ret(json::ValueType::Object);
ret["proposing"] = (mode_.get() == ConsensusMode::Proposing);
ret["proposers"] = static_cast<int>(currPeerPositions_.size());
if (mode_.get() != ConsensusMode::WrongLedger)
{
ret["synched"] = true;
ret["ledger_seq"] = static_cast<std::uint32_t>(previousLedger_.seq()) + 1;
ret["close_granularity"] = static_cast<Int>(closeResolution_.count());
}
else
{
ret["synched"] = false;
}
ret["phase"] = to_string(phase_);
if (result_ && !result_->disputes.empty() && !full)
ret["disputes"] = static_cast<Int>(result_->disputes.size());
if (result_)
ret["our_position"] = result_->position.getJson();
if (full)
{
if (result_)
ret["current_ms"] = static_cast<Int>(result_->roundTime.read().count());
ret["converge_percent"] = convergePercent_;
ret["close_resolution"] = static_cast<Int>(closeResolution_.count());
ret["have_time_consensus"] = haveCloseTimeConsensus_;
ret["previous_proposers"] = static_cast<Int>(prevProposers_);
ret["previous_mseconds"] = static_cast<Int>(prevRoundTime_.count());
if (!currPeerPositions_.empty())
{
json::Value ppj(json::ValueType::Object);
for (auto const& [nodeId, peerPos] : currPeerPositions_)
{
ppj[to_string(nodeId)] = peerPos.getJson();
}
ret["peer_positions"] = std::move(ppj);
}
if (!acquired_.empty())
{
json::Value acq(json::ValueType::Array);
for (auto const& at : acquired_)
{
acq.append(to_string(at.first));
}
ret["acquired"] = std::move(acq);
}
if (result_ && !result_->disputes.empty())
{
json::Value dsj(json::ValueType::Object);
for (auto const& [txId, dispute] : result_->disputes)
{
dsj[to_string(txId)] = dispute.getJson();
}
ret["disputes"] = std::move(dsj);
}
if (!rawCloseTimes_.peers.empty())
{
json::Value ctj(json::ValueType::Object);
for (auto const& ct : rawCloseTimes_.peers)
{
ctj[std::to_string(ct.first.time_since_epoch().count())] = ct.second;
}
ret["close_times"] = std::move(ctj);
}
if (!deadNodes_.empty())
{
json::Value dnj(json::ValueType::Array);
for (auto const& dn : deadNodes_)
{
dnj.append(to_string(dn));
}
ret["dead_nodes"] = std::move(dnj);
}
}
return ret;
}
// Handle a change in the prior ledger during a consensus round
template <class Adaptor>
void
Consensus<Adaptor>::handleWrongLedger(
Ledger_t::ID const& lgrId,
std::unique_ptr<std::stringstream> const& clog)
{
CLOG(clog) << "handleWrongLedger. ";
XRPL_ASSERT(
lgrId != prevLedgerID_ || previousLedger_.id() != lgrId,
"xrpl::Consensus::handleWrongLedger : have wrong ledger");
// Stop proposing because we are out of sync
leaveConsensus(clog);
// First time switching to this ledger
if (prevLedgerID_ != lgrId)
{
prevLedgerID_ = lgrId;
// Clear out state
if (result_)
{
result_->disputes.clear();
result_->compares.clear();
}
currPeerPositions_.clear();
rawCloseTimes_.peers.clear();
deadNodes_.clear();
// Get back in sync, this will also recreate disputes
playbackProposals();
}
if (previousLedger_.id() == prevLedgerID_)
{
CLOG(clog) << "previousLedger_.id() == prevLeverID_ " << prevLedgerID_ << ". ";
return;
}
// we need to switch the ledger we're working from
if (auto newLedger = adaptor_.acquireLedger(prevLedgerID_))
{
JLOG(j_.info()) << "Have the consensus ledger " << prevLedgerID_;
CLOG(clog) << "Have the consensus ledger " << prevLedgerID_ << ". ";
startRoundInternal(
now_,
lgrId,
*newLedger,
ConsensusMode::SwitchedLedger,
clog,
StartRoundReason::Recovered);
}
else
{
CLOG(clog) << "Still on wrong ledger. ";
mode_.set(ConsensusMode::WrongLedger, adaptor_);
}
}
template <class Adaptor>
void
Consensus<Adaptor>::checkLedger(std::unique_ptr<std::stringstream> const& clog)
{
CLOG(clog) << "checkLedger. ";
auto netLgr = adaptor_.getPrevLedger(prevLedgerID_, previousLedger_, mode_.get());
CLOG(clog) << "network ledgerid " << netLgr << ", "
<< "previous ledger " << prevLedgerID_ << ". ";
if (netLgr != prevLedgerID_)
{
std::stringstream ss;
ss << "View of consensus changed during " << to_string(phase_)
<< " mode=" << to_string(mode_.get()) << ", " << prevLedgerID_ << " to " << netLgr
<< ", " << json::Compact{previousLedger_.getJson()} << ". ";
JLOG(j_.warn()) << ss.str();
CLOG(clog) << ss.str();
CLOG(clog) << "State on consensus change " << json::Compact{getJson(true)} << ". ";
handleWrongLedger(netLgr, clog);
}
else if (previousLedger_.id() != prevLedgerID_)
{
CLOG(clog) << "previousLedger_.id() != prevLedgerID_: " << previousLedger_.id() << ','
<< to_string(prevLedgerID_) << ". ";
handleWrongLedger(netLgr, clog);
}
}
template <class Adaptor>
void
Consensus<Adaptor>::playbackProposals()
{
for (auto const& it : recentPeerPositions_)
{
for (auto const& pos : it.second)
{
if (pos.proposal().prevLedger() == prevLedgerID_)
{
if (peerProposalInternal(now_, pos))
adaptor_.share(pos);
}
}
}
}
template <class Adaptor>
void
Consensus<Adaptor>::phaseOpen(std::unique_ptr<std::stringstream> const& clog)
{
CLOG(clog) << "phaseOpen. ";
using namespace std::chrono;
// it is shortly before ledger close time
bool const anyTransactions = adaptor_.hasOpenTransactions();
auto proposersClosed = currPeerPositions_.size();
auto proposersValidated = adaptor_.proposersValidated(prevLedgerID_);
openTime_.tick(clock_.now());
// This computes how long since last ledger's close time
milliseconds sinceClose;
{
auto const mode = mode_.get();
bool const closeAgree = previousLedger_.closeAgree();
auto const prevCloseTime = previousLedger_.closeTime();
auto const prevParentCloseTimePlus1 = previousLedger_.parentCloseTime() + 1s;
bool const previousCloseCorrect = (mode != ConsensusMode::WrongLedger) && closeAgree &&
(prevCloseTime != prevParentCloseTimePlus1);
auto const lastCloseTime = previousCloseCorrect
? prevCloseTime // use consensus timing
: prevCloseTime_; // use the time we saw internally
if (now_ >= lastCloseTime)
{
sinceClose = duration_cast<milliseconds>(now_ - lastCloseTime);
}
else
{
sinceClose = -duration_cast<milliseconds>(lastCloseTime - now_);
}
CLOG(clog) << "calculating how long since last ledger's close time "
"based on mode : "
<< to_string(mode) << ", previous closeAgree: " << closeAgree
<< ", previous close time: " << to_string(prevCloseTime)
<< ", previous parent close time + 1s: " << to_string(prevParentCloseTimePlus1)
<< ", previous close time seen internally: " << to_string(prevCloseTime_)
<< ", last close time: " << to_string(lastCloseTime)
<< ", since close: " << sinceClose.count() << ". ";
}
auto const idleInterval = std::max<milliseconds>(
adaptor_.parms().ledgerIdleInterval, 2 * previousLedger_.closeTimeResolution());
CLOG(clog) << "idle interval set to " << idleInterval.count() << "ms based on "
<< "ledgerIDLE_INTERVAL: " << adaptor_.parms().ledgerIdleInterval.count()
<< ", previous ledger close time resolution: "
<< previousLedger_.closeTimeResolution().count() << "ms. ";
// Decide if we should close the ledger
if (shouldCloseLedger(
anyTransactions,
prevProposers_,
proposersClosed,
proposersValidated,
prevRoundTime_,
sinceClose,
openTime_.read(),
idleInterval,
adaptor_.parms(),
j_,
clog))
{
CLOG(clog) << "closing ledger. ";
closeLedger(clog);
}
}
template <class Adaptor>
bool
Consensus<Adaptor>::shouldPause(std::unique_ptr<std::stringstream> const& clog) const
{
CLOG(clog) << "shouldPause? ";
auto const& parms = adaptor_.parms();
std::uint32_t const ahead(
previousLedger_.seq() - std::min(adaptor_.getValidLedgerIndex(), previousLedger_.seq()));
auto [quorum, trustedKeys] = adaptor_.getQuorumKeys();
std::size_t const totalValidators = trustedKeys.size();
std::size_t const laggards = adaptor_.laggards(previousLedger_.seq(), trustedKeys);
std::size_t const offline = trustedKeys.size();
std::stringstream vars;
vars << " consensuslog (working seq: " << previousLedger_.seq() << ", "
<< "validated seq: " << adaptor_.getValidLedgerIndex() << ", "
<< "am validator: " << adaptor_.validator() << ", "
<< "have validated: " << adaptor_.haveValidated()
<< ", "
// NOLINTBEGIN(bugprone-unchecked-optional-access) result_ is always set when shouldPause
// is called (from phaseEstablish after assert)
<< "roundTime: " << result_->roundTime.read().count()
<< ", "
// NOLINTEND(bugprone-unchecked-optional-access)
<< "max consensus time: " << parms.ledgerMaxConsensus.count() << ", "
<< "validators: " << totalValidators << ", "
<< "laggards: " << laggards << ", "
<< "offline: " << offline << ", "
<< "quorum: " << quorum << ")";
if ((ahead == 0u) || (laggards == 0u) || (totalValidators == 0u) || !adaptor_.validator() ||
!adaptor_.haveValidated() ||
// NOLINTNEXTLINE(bugprone-unchecked-optional-access) result_ set as shouldPause called
result_->roundTime.read() > parms.ledgerMaxConsensus)
{
j_.debug() << "not pausing (early)" << vars.str();
CLOG(clog) << "Not pausing (early). ";
return false;
}
bool willPause = false;
/**
* Maximum phase with distinct thresholds to determine how
* many validators must be on our same ledger sequence number.
* The threshold for the 1st (0) phase is >= the minimum number that
* can achieve quorum. Threshold for the maximum phase is 100%
* of all trusted validators. Progression from min to max phase is
* simply linear. If there are 5 phases (maxPausePhase = 4)
* and minimum quorum is 80%, then thresholds progress as follows:
* 0: >=80%
* 1: >=85%
* 2: >=90%
* 3: >=95%
* 4: =100%
*/
static constexpr std::size_t kMaxPausePhase = 4;
/**
* No particular threshold guarantees consensus. Lower thresholds
* are easier to achieve than higher, but higher thresholds are
* more likely to reach consensus. Cycle through the phases if
* lack of synchronization continues.
*
* Current information indicates that no phase is likely to be intrinsically
* better than any other: the lower the threshold, the less likely that
* up-to-date nodes will be able to reach consensus without the laggards.
* But the higher the threshold, the longer the likely resulting pause.
* 100% is slightly less desirable in the long run because the potential
* of a single dawdling peer to slow down everything else. So if we
* accept that no phase is any better than any other phase, but that
* all of them will potentially enable us to arrive at consensus, cycling
* through them seems to be appropriate. Further, if we do reach the
* point of having to cycle back around, then it's likely that something
* else out of the scope of this delay mechanism is wrong with the
* network.
*/
std::size_t const phase = (ahead - 1) % (kMaxPausePhase + 1);
// validators that remain after the laggards() function are considered
// offline, and should be considered as laggards for purposes of
// evaluating whether the threshold for non-laggards has been reached.
switch (phase)
{
case 0:
// Laggards and offline shouldn't preclude consensus.
if (laggards + offline > totalValidators - quorum)
willPause = true;
break;
case kMaxPausePhase:
// No tolerance.
willPause = true;
break;
default:
// Ensure that sufficient validators are known to be not lagging.
// Their sufficiently most recent validation sequence was equal to
// or greater than our own.
//
// The threshold is the amount required for quorum plus
// the proportion of the remainder based on number of intermediate
// phases between 0 and max.
float const nonLaggards = totalValidators - (laggards + offline);
float const quorumRatio = static_cast<float>(quorum) / totalValidators;
float const allowedDissent = 1.0f - quorumRatio;
float const phaseFactor = static_cast<float>(phase) / kMaxPausePhase;
if (nonLaggards / totalValidators < quorumRatio + (allowedDissent * phaseFactor))
{
willPause = true;
}
}
if (willPause)
{
j_.warn() << "pausing" << vars.str();
CLOG(clog) << "pausing " << vars.str() << ". ";
}
else
{
j_.debug() << "not pausing" << vars.str();
CLOG(clog) << "not pausing. ";
}
return willPause;
}
template <class Adaptor>
void
Consensus<Adaptor>::phaseEstablish(std::unique_ptr<std::stringstream> const& clog)
{
CLOG(clog) << "phaseEstablish. ";
// can only establish consensus if we already took a stance
XRPL_ASSERT(result_, "xrpl::Consensus::phaseEstablish : result is set");
// NOLINTBEGIN(bugprone-unchecked-optional-access) assert above
startEstablishTracing();
++peerUnchangedCounter_;
++establishCounter_;
using namespace std::chrono;
ConsensusParms const& parms = adaptor_.parms();
result_->roundTime.tick(clock_.now());
result_->proposers = currPeerPositions_.size();
convergePercent_ = result_->roundTime.read() * 100 /
std::max<milliseconds>(prevRoundTime_, parms.avMinConsensusTime);
CLOG(clog) << "convergePercent_ " << convergePercent_
<< " is based on round duration so far: " << result_->roundTime.read().count()
<< "ms, "
<< "previous round duration: " << prevRoundTime_.count() << "ms, "
<< "avMIN_CONSENSUS_TIME: " << parms.avMinConsensusTime.count() << "ms. ";
// Give everyone a chance to take an initial position
if (result_->roundTime.read() < parms.ledgerMinConsensus)
{
CLOG(clog) << "ledgerMIN_CONSENSUS not reached: " << parms.ledgerMinConsensus.count()
<< "ms. ";
return;
}
updateOurPositions(clog);
updateEstablishTracing();
// Nothing to do if too many laggards or we don't have consensus.
if (shouldPause(clog) || !haveConsensus(clog))
return;
if (!haveCloseTimeConsensus_)
{
JLOG(j_.info()) << "We have TX consensus but not CT consensus";
CLOG(clog) << "We have TX consensus but not CT consensus. ";
return;
}
JLOG(j_.info()) << "Converge cutoff (" << currPeerPositions_.size() << " participants)";
CLOG(clog) << "Converge cutoff (" << currPeerPositions_.size()
<< " participants). Transitioned to ConsensusPhase::Accepted. ";
adaptor_.updateOperatingMode(currPeerPositions_.size());
prevProposers_ = currPeerPositions_.size();
prevRoundTime_ = result_->roundTime.read();
endEstablishTracing();
{
namespace cs = telemetry::consensus::span;
if (result_->state == ConsensusState::Yes)
{
adaptor_.onOutcomeEvent(cs::event::outcomeYes);
}
else if (result_->state == ConsensusState::MovedOn)
{
adaptor_.onOutcomeEvent(cs::event::outcomeMovedOn);
}
else if (result_->state == ConsensusState::Expired)
{
adaptor_.onOutcomeEvent(cs::event::outcomeExpired);
}
}
phase_ = ConsensusPhase::Accepted;
adaptor_.onPhaseEvent(
telemetry::consensus::span::event::phaseAccepted,
telemetry::consensus::span::val::phaseAccepted);
JLOG(j_.debug()) << "transitioned to ConsensusPhase::Accepted";
adaptor_.onAccept(
*result_,
previousLedger_,
closeResolution_,
rawCloseTimes_,
mode_.get(),
getJson(true),
adaptor_.validating());
// NOLINTEND(bugprone-unchecked-optional-access)
}
template <class Adaptor>
void
Consensus<Adaptor>::closeLedger(std::unique_ptr<std::stringstream> const& clog)
{
// We should not be closing if we already have a position
XRPL_ASSERT(!result_, "xrpl::Consensus::closeLedger : result is not set");
// Annotate the open-phase span with end-of-phase metadata before
// ending it, so a Tempo/Jaeger query for consensus.phase.open shows
// how long the phase ran and how many peer positions arrived during it.
openTime_.tick(clock_.now());
if (openSpan_ && *openSpan_)
{
namespace cs = telemetry::consensus::span;
openSpan_->setAttribute(
cs::attr::openDurationMs, static_cast<int64_t>(openTime_.read().count()));
openSpan_->setAttribute(
cs::attr::peerPositionsAtClose, static_cast<int64_t>(currPeerPositions_.size()));
}
openSpan_.reset();
phase_ = ConsensusPhase::Establish;
adaptor_.onPhaseEvent(
telemetry::consensus::span::event::phaseEstablish,
telemetry::consensus::span::val::phaseEstablish);
JLOG(j_.debug()) << "transitioned to ConsensusPhase::Establish";
rawCloseTimes_.self = now_;
peerUnchangedCounter_ = 0;
establishCounter_ = 0;
result_.emplace(adaptor_.onClose(previousLedger_, now_, mode_.get()));
result_->roundTime.reset(clock_.now());
// Share the newly created transaction set if we haven't already
// received it from a peer
if (acquired_.emplace(result_->txns.id(), result_->txns).second)
adaptor_.share(result_->txns);
auto const mode = mode_.get();
CLOG(clog) << "closeLedger transitioned to ConsensusPhase::Establish, mode: " << to_string(mode)
<< ", number of peer positions: " << currPeerPositions_.size() << ". ";
if (mode == ConsensusMode::Proposing)
adaptor_.propose(result_->position);
// Create disputes with any peer positions we have transactions for
for (auto const& pit : currPeerPositions_)
{
auto const& pos = pit.second.proposal().position();
auto const it = acquired_.find(pos);
if (it != acquired_.end())
createDisputes(it->second, clog);
}
}
/**
* How many of the participants must agree to reach a given threshold?
*
* Note that the number may not precisely yield the requested percentage.
* For example, with with size = 5 and percent = 70, we return 3, but
* 3 out of 5 works out to 60%. There are no security implications to
* this.
*
* @param participants The number of participants (i.e. validators)
* @param percent The percent that we want to reach
*
* @return the number of participants which must agree
*/
inline int
participantsNeeded(int participants, int percent)
{
int const result = ((participants * percent) + (percent / 2)) / 100;
return (result == 0) ? 1 : result;
}
template <class Adaptor>
void
Consensus<Adaptor>::updateOurPositions(std::unique_ptr<std::stringstream> const& clog)
{
// We must have a position if we are updating it
XRPL_ASSERT(result_, "xrpl::Consensus::updateOurPositions : result is set");
// NOLINTBEGIN(bugprone-unchecked-optional-access) assert above
using namespace telemetry;
// Child of the establish span via its captured context (establishSpan_ is
// a thread-free SpanGuard, so parent explicitly via its context). Null
// context (establish not started) yields a null guard, same as before.
auto span = SpanGuard::childSpan(consensus::span::updatePositions, establishSpanContext_);
span.setAttribute(
consensus::span::attr::convergePercent, static_cast<int64_t>(convergePercent_));
span.setAttribute(
consensus::span::attr::proposers, static_cast<int64_t>(currPeerPositions_.size()));
span.setAttribute(
consensus::span::attr::disputesCount, static_cast<int64_t>(result_->disputes.size()));
ConsensusParms const& parms = adaptor_.parms();
// Compute a cutoff time
auto const peerCutoff = now_ - parms.proposeFRESHNESS;
auto const ourCutoff = now_ - parms.proposeINTERVAL;
CLOG(clog) << "updateOurPositions. peerCutoff " << to_string(peerCutoff) << ", ourCutoff "
<< to_string(ourCutoff) << ". ";
// Verify freshness of peer positions and compute close times
std::map<NetClock::time_point, int> closeTimeVotes;
{
auto it = currPeerPositions_.begin();
while (it != currPeerPositions_.end())
{
Proposal_t const& peerProp = it->second.proposal();
if (peerProp.isStale(peerCutoff))
{
// peer's proposal is stale, so remove it
NodeID_t const& peerID = peerProp.nodeID();
JLOG(j_.warn()) << "Removing stale proposal from " << peerID;
for (auto& dt : result_->disputes)
dt.second.unVote(peerID);
it = currPeerPositions_.erase(it);
}
else
{
// proposal is still fresh
++closeTimeVotes[asCloseTime(peerProp.closeTime())];
++it;
}
}
}
// This will stay unseated unless there are any changes
std::optional<TxSet_t> ourNewSet;
// Update votes on disputed transactions
{
std::optional<typename TxSet_t::MutableTxSet> mutableSet;
for (auto& [txId, dispute] : result_->disputes)
{
// Because the threshold for inclusion increases,
// time can change our position on a dispute
if (dispute.updateVote(
convergePercent_, mode_.get() == ConsensusMode::Proposing, parms))
{
if (!mutableSet)
mutableSet.emplace(result_->txns);
if (dispute.getOurVote())
{
// now a yes
mutableSet->insert(dispute.tx());
}
else
{
// now a no
mutableSet->erase(txId);
}
auto const yaysStr = std::to_string(dispute.getYays());
auto const naysStr = std::to_string(dispute.getNays());
span.addEvent(
consensus::span::event::disputeResolve,
{{consensus::span::attr::txId, to_string(txId)},
{consensus::span::attr::disputeOurVote,
dispute.getOurVote() ? std::string_view{consensus::span::val::yes}
: std::string_view{consensus::span::val::no}},
{consensus::span::attr::disputeYays, yaysStr},
{consensus::span::attr::disputeNays, naysStr}});
}
}
if (mutableSet)
ourNewSet.emplace(std::move(*mutableSet));
}
NetClock::time_point consensusCloseTime = {};
haveCloseTimeConsensus_ = false;
if (currPeerPositions_.empty())
{
// no other times
haveCloseTimeConsensus_ = true;
consensusCloseTime = asCloseTime(result_->position.closeTime());
}
else
{
// We don't track rounds for close time, so just pass 0s
auto const [neededWeight, newState] =
getNeededWeight(parms, closeTimeAvalancheState_, convergePercent_, 0, 0);
if (newState)
closeTimeAvalancheState_ = *newState;
CLOG(clog) << "neededWeight " << neededWeight << ". ";
span.setAttribute(
consensus::span::attr::avalancheThreshold, static_cast<int64_t>(neededWeight));
int participants = currPeerPositions_.size();
if (mode_.get() == ConsensusMode::Proposing)
{
++closeTimeVotes[asCloseTime(result_->position.closeTime())];
++participants;
}
// Threshold for non-zero vote
int threshVote = participantsNeeded(participants, neededWeight);
// Threshold to declare consensus
int const threshConsensus = participantsNeeded(participants, parms.avCtConsensusPct);
std::stringstream ss;
ss << "Proposers:" << currPeerPositions_.size() << " nw:" << neededWeight
<< " thrV:" << threshVote << " thrC:" << threshConsensus;
JLOG(j_.info()) << ss.str();
CLOG(clog) << ss.str();
for (auto const& [t, v] : closeTimeVotes)
{
JLOG(j_.debug()) << "CCTime: seq "
<< static_cast<std::uint32_t>(previousLedger_.seq()) + 1 << ": "
<< t.time_since_epoch().count() << " has " << v << ", " << threshVote
<< " required";
if (v >= threshVote)
{
// A close time has enough votes for us to try to agree
consensusCloseTime = t;
threshVote = v;
if (threshVote >= threshConsensus)
haveCloseTimeConsensus_ = true;
}
}
if (!haveCloseTimeConsensus_)
{
JLOG(j_.debug()) << "No CT consensus:"
<< " Proposers:" << currPeerPositions_.size()
<< " Mode:" << to_string(mode_.get()) << " Thresh:" << threshConsensus
<< " Pos:" << consensusCloseTime.time_since_epoch().count();
CLOG(clog) << "No close time consensus. ";
}
}
span.setAttribute(consensus::span::attr::haveCloseTimeConsensus, haveCloseTimeConsensus_);
span.setAttribute(
consensus::span::attr::closeTimeThreshold, static_cast<int64_t>(parms.avCtConsensusPct));
if (!ourNewSet &&
((consensusCloseTime != asCloseTime(result_->position.closeTime())) ||
result_->position.isStale(ourCutoff)))
{
// close time changed or our position is stale
ourNewSet.emplace(result_->txns);
}
if (ourNewSet)
{
auto newID = ourNewSet->id();
result_->txns = std::move(*ourNewSet);
std::stringstream ss;
ss << "Position change: CTime " << consensusCloseTime.time_since_epoch().count() << ", tx "
<< newID;
JLOG(j_.info()) << ss.str();
CLOG(clog) << ss.str();
result_->position.changePosition(newID, consensusCloseTime, now_);
// Share our new transaction set and update disputes
// if we haven't already received it
if (acquired_.emplace(newID, result_->txns).second)
{
if (!result_->position.isBowOut())
adaptor_.share(result_->txns);
for (auto const& [nodeId, peerPos] : currPeerPositions_)
{
Proposal_t const& p = peerPos.proposal();
if (p.position() == newID)
updateDisputes(nodeId, result_->txns);
}
}
// Share our new position if we are still participating this round
if (!result_->position.isBowOut() && (mode_.get() == ConsensusMode::Proposing))
adaptor_.propose(result_->position);
}
// NOLINTEND(bugprone-unchecked-optional-access)
}
template <class Adaptor>
bool
Consensus<Adaptor>::haveConsensus(std::unique_ptr<std::stringstream> const& clog)
{
// Must have a stance if we are checking for consensus
XRPL_ASSERT(result_, "xrpl::Consensus::haveConsensus : has result");
// NOLINTBEGIN(bugprone-unchecked-optional-access) assert above
using namespace telemetry;
// Child of the establish span via its captured context (establishSpan_ is
// a thread-free SpanGuard, so parent explicitly via its context).
auto span = SpanGuard::childSpan(consensus::span::check, establishSpanContext_);
// CHECKME: should possibly count unacquired TX sets as disagreeing
int agree = 0, disagree = 0;
auto ourPosition = result_->position.position();
// Count number of agreements/disagreements with our position
for (auto const& [nodeId, peerPos] : currPeerPositions_)
{
Proposal_t const& peerProp = peerPos.proposal();
if (peerProp.position() == ourPosition)
{
++agree;
}
else
{
JLOG(j_.debug()) << "Proposal disagreement: Peer " << nodeId << " has "
<< peerProp.position();
++disagree;
}
}
auto currentFinished = adaptor_.proposersFinished(previousLedger_, prevLedgerID_);
JLOG(j_.debug()) << "Checking for TX consensus: agree=" << agree << ", disagree=" << disagree;
ConsensusParms const& parms = adaptor_.parms();
// Stalling is BAD. It means that we have a consensus on the close time, so
// peers are talking, but we have disputed transactions that peers are
// unable or unwilling to come to agreement on one way or the other.
bool const stalled =
haveCloseTimeConsensus_ && !result_->disputes.empty() &&
std::ranges::all_of(result_->disputes, [this, &parms, &clog](auto const& dispute) {
return dispute.second.stalled(
parms, mode_.get() == ConsensusMode::Proposing, peerUnchangedCounter_, j_, clog);
});
if (stalled)
{
std::stringstream ss;
ss << "Consensus detects as stalled with " << (agree + disagree) << "/" << prevProposers_
<< " proposers, and " << result_->disputes.size() << " stalled disputed transactions.";
JLOG(j_.error()) << ss.str();
CLOG(clog) << ss.str();
}
// Determine if we actually have consensus or not
result_->state = checkConsensus(
prevProposers_,
agree + disagree,
agree,
currentFinished,
prevRoundTime_,
result_->roundTime.read(),
stalled,
parms,
mode_.get() == ConsensusMode::Proposing,
j_,
clog);
// Set span attributes before the early-return branches below so the
// consensus.check span carries diagnostic data even when consensus is
// not reached (the No / Expired paths return early).
span.setAttribute(consensus::span::attr::agreeCount, static_cast<int64_t>(agree));
span.setAttribute(consensus::span::attr::disagreeCount, static_cast<int64_t>(disagree));
span.setAttribute(
consensus::span::attr::convergePercent, static_cast<int64_t>(convergePercent_));
span.setAttribute(consensus::span::attr::haveCloseTimeConsensus, haveCloseTimeConsensus_);
span.setAttribute(
consensus::span::attr::thresholdPercent,
static_cast<int64_t>(adaptor_.parms().avCtConsensusPct));
span.setAttribute(
consensus::span::attr::proposersFinished, static_cast<int64_t>(currentFinished));
span.setAttribute(consensus::span::attr::consensusStalled, stalled);
span.setAttribute(
consensus::span::attr::establishCount, static_cast<int64_t>(establishCounter_));
std::string_view stateStr = consensus::span::val::no;
if (result_->state == ConsensusState::Yes)
{
stateStr = consensus::span::val::yes;
}
else if (result_->state == ConsensusState::MovedOn)
{
stateStr = consensus::span::val::movedOn;
}
else if (result_->state == ConsensusState::Expired)
{
stateStr = consensus::span::val::expired;
}
span.setAttribute(consensus::span::attr::consensusResult, stateStr);
if (result_->state == ConsensusState::No)
{
CLOG(clog) << "No consensus. ";
return false;
}
// Consensus has taken far too long. Drop out of the round.
if (result_->state == ConsensusState::Expired)
{
static auto const kMinimumCounter = parms.avalancheCutoffs.size() * parms.avMinRounds;
std::stringstream ss;
if (establishCounter_ < kMinimumCounter)
{
// If each round of phaseEstablish takes a very long time, we may
// "expire" before we've given consensus enough time at each
// avalanche level to actually come to a consensus. In that case,
// keep trying. This should only happen if there are an extremely
// large number of disputes such that each round takes an inordinate
// amount of time.
ss << "Consensus time has expired in round " << establishCounter_
<< "; continue until round " << kMinimumCounter << ". "
<< json::Compact{getJson(false)};
JLOG(j_.error()) << ss.str();
CLOG(clog) << ss.str() << ". ";
return false;
}
ss << "Consensus expired. " << json::Compact{getJson(true)};
JLOG(j_.error()) << ss.str();
CLOG(clog) << ss.str() << ". ";
leaveConsensus(clog);
}
// There is consensus, but we need to track if the network moved on
// without us.
if (result_->state == ConsensusState::MovedOn)
{
JLOG(j_.error()) << "Unable to reach consensus";
JLOG(j_.error()) << json::Compact{getJson(true)};
CLOG(clog) << "Unable to reach consensus " << json::Compact{getJson(true)} << ". ";
}
CLOG(clog) << "Consensus has been reached. ";
// NOLINTEND(bugprone-unchecked-optional-access)
return true;
}
template <class Adaptor>
void
Consensus<Adaptor>::leaveConsensus(std::unique_ptr<std::stringstream> const& clog)
{
if (mode_.get() == ConsensusMode::Proposing)
{
if (result_ && !result_->position.isBowOut())
{
result_->position.bowOut(now_);
adaptor_.propose(result_->position);
}
mode_.set(ConsensusMode::Observing, adaptor_);
JLOG(j_.info()) << "Bowing out of consensus";
CLOG(clog) << "Bowing out of consensus. ";
}
}
template <class Adaptor>
void
Consensus<Adaptor>::createDisputes(TxSet_t const& o, std::unique_ptr<std::stringstream> const& clog)
{
// Cannot create disputes without our stance
XRPL_ASSERT(result_, "xrpl::Consensus::createDisputes : result is set");
// NOLINTBEGIN(bugprone-unchecked-optional-access) assert above
// Only create disputes if this is a new set
auto const emplaced = result_->compares.emplace(o.id()).second;
CLOG(clog) << "createDisputes: new set? " << !emplaced << ". ";
if (!emplaced)
return;
// Nothing to dispute if we agree
if (result_->txns.id() == o.id())
{
CLOG(clog) << "both sets are identical. ";
return;
}
CLOG(clog) << "comparing existing with new set: " << result_->txns.id() << ',' << o.id()
<< ". ";
JLOG(j_.debug()) << "createDisputes " << result_->txns.id() << " to " << o.id();
auto differences = result_->txns.compare(o);
int dc = 0;
for (auto const& [txId, inThisSet] : differences)
{
++dc;
// create disputed transactions (from the ledger that has them)
XRPL_ASSERT(
(inThisSet && result_->txns.find(txId) && !o.find(txId)) ||
(!inThisSet && !result_->txns.find(txId) && o.find(txId)),
"xrpl::Consensus::createDisputes : has disputed transactions");
Tx_t const tx = inThisSet ? result_->txns.find(txId) : o.find(txId);
auto txID = tx.id();
if (result_->disputes.find(txID) != result_->disputes.end())
continue;
JLOG(j_.debug()) << "Transaction " << txID << " is disputed";
typename Result::Dispute_t dtx{
tx,
result_->txns.exists(txID),
std::max(prevProposers_, currPeerPositions_.size()),
j_};
// Update all of the available peer's votes on the disputed transaction
for (auto const& [nodeId, peerPos] : currPeerPositions_)
{
Proposal_t const& peerProp = peerPos.proposal();
auto const cit = acquired_.find(peerProp.position());
if (cit != acquired_.end() && dtx.setVote(nodeId, cit->second.exists(txID)))
peerUnchangedCounter_ = 0;
}
adaptor_.share(dtx.tx());
result_->disputes.emplace(txID, std::move(dtx));
}
JLOG(j_.debug()) << dc << " differences found";
CLOG(clog) << "disputes: " << dc << ". ";
// NOLINTEND(bugprone-unchecked-optional-access)
}
template <class Adaptor>
void
Consensus<Adaptor>::updateDisputes(NodeID_t const& node, TxSet_t const& other)
{
// Cannot updateDisputes without our stance
XRPL_ASSERT(result_, "xrpl::Consensus::updateDisputes : result is set");
// NOLINTBEGIN(bugprone-unchecked-optional-access) assert above
// Ensure we have created disputes against this set if we haven't seen
// it before
if (result_->compares.find(other.id()) == result_->compares.end())
createDisputes(other);
for (auto& it : result_->disputes)
{
auto& d = it.second;
if (d.setVote(node, other.exists(d.tx().id())))
peerUnchangedCounter_ = 0;
}
// NOLINTEND(bugprone-unchecked-optional-access)
}
template <class Adaptor>
NetClock::time_point
Consensus<Adaptor>::asCloseTime(NetClock::time_point raw) const
{
return roundCloseTime(raw, closeResolution_);
}
template <class Adaptor>
void
Consensus<Adaptor>::startEstablishTracing()
{
if (establishSpan_)
return;
// Child of the round span via its captured context: parent establish
// explicitly under roundSpanContext_. An invalid round context (round span
// not yet created) yields a null guard.
establishSpan_.emplace(
telemetry::SpanGuard::childSpan(
telemetry::consensus::span::establish, adaptor_.roundSpanContext()));
// Capture the establish context; children (update_positions, check) parent
// to it explicitly via establishSpanContext_. establishSpan_ is a
// thread-free SpanGuard reset() on a different worker than it is emplaced
// on -- no scope work.
if (*establishSpan_)
{
establishSpanContext_ = establishSpan_->spanContext();
}
}
template <class Adaptor>
void
Consensus<Adaptor>::updateEstablishTracing()
{
if (!establishSpan_)
return;
namespace cs = telemetry::consensus::span;
establishSpan_->setAttribute(cs::attr::convergePercent, static_cast<int64_t>(convergePercent_));
establishSpan_->setAttribute(cs::attr::establishCount, static_cast<int64_t>(establishCounter_));
establishSpan_->setAttribute(
cs::attr::proposers, static_cast<int64_t>(currPeerPositions_.size()));
if (result_)
{
establishSpan_->setAttribute(
cs::attr::disputesCount, static_cast<int64_t>(result_->disputes.size()));
}
}
template <class Adaptor>
void
Consensus<Adaptor>::endEstablishTracing()
{
establishSpan_.reset();
establishSpanContext_ = telemetry::SpanContext{};
}
} // namespace xrpl
Reference in New Issue View Git Blame Copy Permalink