Merge branch 'pratik/otel-phase2-rpc-tracing' into pratik/otel-phase3-tx-tracing

Signed-off-by: Pratik Mankawde <3397372+pratikmankawde@users.noreply.github.com>
This commit is contained in:
Pratik Mankawde
2026-07-27 16:58:48 +01:00
251 changed files with 9348 additions and 7803 deletions

View File

@@ -5,6 +5,7 @@
#include <xrpl/beast/container/detail/aged_associative_container.h>
#include <xrpl/beast/container/detail/aged_container_iterator.h>
#include <xrpl/beast/container/detail/empty_base_optimization.h>
#include <xrpl/beast/utility/instrumentation.h>
#include <boost/intrusive/list.hpp>
#include <boost/intrusive/unordered_set.hpp>

View File

@@ -0,0 +1,126 @@
#pragma once
#include <xrpl/basics/algorithm.h>
#include <algorithm>
#include <functional>
#include <utility>
#include <vector>
namespace xrpl {
template <class TxID, class Sequence>
class CensorshipDetector
{
public:
struct TxIDSeq
{
TxID txid;
Sequence seq;
TxIDSeq(TxID const& txid, Sequence const& seq) : txid(txid), seq(seq)
{
}
};
friend bool
operator<(TxIDSeq const& lhs, TxIDSeq const& rhs)
{
if (lhs.txid != rhs.txid)
return lhs.txid < rhs.txid;
return lhs.seq < rhs.seq;
}
friend bool
operator<(TxIDSeq const& lhs, TxID const& rhs)
{
return lhs.txid < rhs;
}
friend bool
operator<(TxID const& lhs, TxIDSeq const& rhs)
{
return lhs < rhs.txid;
}
using TxIDSeqVec = std::vector<TxIDSeq>;
private:
TxIDSeqVec tracker_;
public:
CensorshipDetector() = default;
/**
* Add transactions being proposed for the current consensus round.
*
* @param proposed The set of transactions that we are initially proposing
* for this round.
*/
void
propose(TxIDSeqVec proposed)
{
// We want to remove any entries that we proposed in a previous round
// that did not make it in yet if we are no longer proposing them.
// And we also want to preserve the Sequence of entries that we proposed
// in the last round and want to propose again.
std::sort(proposed.begin(), proposed.end());
generalizedSetIntersection(
proposed.begin(),
proposed.end(),
tracker_.cbegin(),
tracker_.cend(),
[](auto& x, auto const& y) { x.seq = y.seq; },
[](auto const& x, auto const& y) { return x.txid < y.txid; });
tracker_ = std::move(proposed);
}
/**
* Determine which transactions made it and perform censorship detection.
*
* This function is called when the server is proposing and a consensus
* round it participated in completed.
*
* @param accepted The set of transactions that the network agreed
* should be included in the ledger being built.
* @param pred A predicate invoked for every transaction we've proposed
* but which hasn't yet made it. The predicate must be
* callable as:
* bool pred(TxID const&, Sequence)
* It must return true for entries that should be removed.
*/
template <class Predicate>
void
check(std::vector<TxID> accepted, Predicate&& pred)
{
auto acceptTxid = accepted.begin();
auto const ae = accepted.end();
std::sort(acceptTxid, ae);
// We want to remove all tracking entries for transactions that were
// accepted as well as those which match the predicate.
auto i = removeIfIntersectOrMatch(
tracker_.begin(),
tracker_.end(),
accepted.begin(),
accepted.end(),
[&pred](auto const& x) { return pred(x.txid, x.seq); },
std::less<void>{});
tracker_.erase(i, tracker_.end());
}
/**
* Removes all elements from the tracker
*
* Typically, this function might be called after we reconnect to the
* network following an outage, or after we start tracking the network.
*/
void
reset()
{
tracker_.clear();
}
};
} // namespace xrpl

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,210 @@
#pragma once
#include <xrpl/beast/utility/instrumentation.h>
#include <chrono>
#include <cstddef>
#include <map>
#include <optional>
#include <utility>
namespace xrpl {
/**
* Consensus algorithm parameters
*
* Parameters which control the consensus algorithm. This are not
* meant to be changed arbitrarily.
*/
struct ConsensusParms
{
explicit ConsensusParms() = default;
//-------------------------------------------------------------------------
// Validation and proposal durations are relative to NetClock times, so use
// second resolution
/**
* The duration a validation remains current after its ledger's
* close time.
*
* This is a safety to protect against very old validations and the time
* it takes to adjust the close time accuracy window.
*/
std::chrono::seconds const validationValidWall = std::chrono::minutes{5};
/**
* Duration a validation remains current after first observed.
*
* The duration a validation remains current after the time we
* first saw it. This provides faster recovery in very rare cases where the
* number of validations produced by the network is lower than normal
*/
std::chrono::seconds const validationValidLocal = std::chrono::minutes{3};
/**
* Duration pre-close in which validations are acceptable.
*
* The number of seconds before a close time that we consider a validation
* acceptable. This protects against extreme clock errors
*/
std::chrono::seconds const validationValidEarly = std::chrono::minutes{3};
/**
* How long we consider a proposal fresh
*/
std::chrono::seconds const proposeFRESHNESS = std::chrono::seconds{20};
/**
* How often we force generating a new proposal to keep ours fresh
*/
std::chrono::seconds const proposeINTERVAL = std::chrono::seconds{12};
//-------------------------------------------------------------------------
// Consensus durations are relative to the internal Consensus clock and use
// millisecond resolution.
/**
* The percentage threshold above which we can declare consensus.
*/
std::size_t const minConsensusPct = 80;
/**
* The duration a ledger may remain idle before closing
*/
std::chrono::milliseconds const ledgerIdleInterval = std::chrono::seconds{15};
/**
* The number of seconds we wait minimum to ensure participation
*/
std::chrono::milliseconds const ledgerMinConsensus = std::chrono::milliseconds{1950};
/**
* The maximum amount of time to spend pausing for laggards.
*
* This should be sufficiently less than validationFRESHNESS so that
* validators don't appear to be offline that are merely waiting for
* laggards.
*/
std::chrono::milliseconds const ledgerMaxConsensus = std::chrono::seconds{15};
/**
* Minimum number of seconds to wait to ensure others have computed the LCL
*/
std::chrono::milliseconds const ledgerMinClose = std::chrono::seconds{2};
/**
* How often we check state or change positions
*/
std::chrono::milliseconds const ledgerGRANULARITY = std::chrono::seconds{1};
/**
* How long to wait before completely abandoning consensus
*/
std::size_t const ledgerAbandonConsensusFactor = 10;
/**
* Maximum amount of time to give a consensus round
*
* Does not include the time to build the LCL, so there is no reason for a
* round to go this long, regardless of how big the ledger is.
*/
std::chrono::milliseconds const ledgerAbandonConsensus = std::chrono::seconds{120};
/**
* The minimum amount of time to consider the previous round
* to have taken.
*
* The minimum amount of time to consider the previous round
* to have taken. This ensures that there is an opportunity
* for a round at each avalanche threshold even if the
* previous consensus was very fast. This should be at least
* twice the interval between proposals (0.7s) divided by
* the interval between mid and late consensus ([85-50]/100).
*/
std::chrono::milliseconds const avMinConsensusTime = std::chrono::seconds{5};
//------------------------------------------------------------------------------
// Avalanche tuning
// As a function of the percent this round's duration is of the prior round,
// we increase the threshold for yes votes to add a transaction to our
// position.
enum class AvalancheState { Init, Mid, Late, Stuck };
struct AvalancheCutoff
{
int const consensusTime;
std::size_t const consensusPct;
AvalancheState const next;
};
/**
* Map the consensus requirement avalanche state to the amount of time that
* must pass before moving to that state, the agreement percentage required
* at that state, and the next state. "stuck" loops back on itself because
* once we're stuck, we're stuck.
* This structure allows for "looping" of states if needed.
*/
std::map<AvalancheState, AvalancheCutoff> const avalancheCutoffs{
// {state, {time, percent, nextState}},
// Initial state: 50% of nodes must vote yes
{AvalancheState::Init,
{.consensusTime = 0, .consensusPct = 50, .next = AvalancheState::Mid}},
// mid-consensus starts after 50% of the previous round time, and
// requires 65% yes
{AvalancheState::Mid,
{.consensusTime = 50, .consensusPct = 65, .next = AvalancheState::Late}},
// late consensus starts after 85% time, and requires 70% yes
{AvalancheState::Late,
{.consensusTime = 85, .consensusPct = 70, .next = AvalancheState::Stuck}},
// we're stuck after 2x time, requires 95% yes votes
{AvalancheState::Stuck,
{.consensusTime = 200, .consensusPct = 95, .next = AvalancheState::Stuck}},
};
/**
* Percentage of nodes required to reach agreement on ledger close time
*/
std::size_t const avCtConsensusPct = 75;
/**
* Number of rounds before certain actions can happen.
*/
// (Moving to the next avalanche level, considering that votes are stalled
// without consensus.)
std::size_t const avMinRounds = 2;
/**
* Number of rounds before a stuck vote is considered unlikely to change
* because voting stalled
*/
std::size_t const avStalledRounds = 4;
};
inline std::pair<std::size_t, std::optional<ConsensusParms::AvalancheState>>
getNeededWeight(
ConsensusParms const& p,
ConsensusParms::AvalancheState currentState,
int percentTime,
std::size_t currentRounds,
std::size_t minimumRounds)
{
// at() can throw, but the map is built by hand to ensure all valid
// values are available.
auto const& currentCutoff = p.avalancheCutoffs.at(currentState);
// Should we consider moving to the next state?
if (currentCutoff.next != currentState && currentRounds >= minimumRounds)
{
// at() can throw, but the map is built by hand to ensure all
// valid values are available.
auto const& nextCutoff = p.avalancheCutoffs.at(currentCutoff.next);
// See if enough time has passed to move on to the next.
XRPL_ASSERT(
nextCutoff.consensusTime >= currentCutoff.consensusTime,
"xrpl::getNeededWeight : next state valid");
if (percentTime >= nextCutoff.consensusTime)
{
return {nextCutoff.consensusPct, currentCutoff.next};
}
}
return {currentCutoff.consensusPct, {}};
}
} // namespace xrpl

View File

@@ -0,0 +1,299 @@
#pragma once
#include <xrpl/basics/base_uint.h>
#include <xrpl/basics/chrono.h>
#include <xrpl/json/json_value.h>
#include <xrpl/protocol/HashPrefix.h>
#include <xrpl/protocol/digest.h>
#include <xrpl/protocol/jss.h>
#include <cstdint>
#include <optional>
#include <sstream>
#include <string>
namespace xrpl {
/**
* Represents a proposed position taken during a round of consensus.
*
* During consensus, peers seek agreement on a set of transactions to
* apply to the prior ledger to generate the next ledger. Each peer takes a
* position on whether to include or exclude potential transactions.
* The position on the set of transactions is proposed to its peers as an
* instance of the ConsensusProposal class.
*
* An instance of ConsensusProposal can be either our own proposal or one of
* our peer's.
*
* As consensus proceeds, peers may change their position on the transaction,
* or choose to abstain. Each successive proposal includes a strictly
* monotonically increasing number (or, if a peer is choosing to abstain,
* the special value `kSeqLeave`).
*
* Refer to @ref Consensus for requirements of the template arguments.
*
* @tparam NodeId Type used to uniquely identify nodes/peers
* @tparam LedgerId Type used to uniquely identify ledgers
* @tparam Position Type used to represent the position taken on transactions
* under consideration during this round of consensus
*/
template <class NodeId, class LedgerId, class Position>
class ConsensusProposal
{
public:
using NodeID = NodeId;
//< Sequence value when a peer initially joins consensus
static std::uint32_t const kSeqJoin = 0;
//< Sequence number when a peer wants to bow out and leave consensus
static std::uint32_t const kSeqLeave = 0xffffffff;
/**
* Constructor
*
* @param prevLedger The previous ledger this proposal is building on.
* @param seq The sequence number of this proposal.
* @param position The position taken on transactions in this round.
* @param closeTime Position of when this ledger closed.
* @param now Time when the proposal was taken.
* @param nodeID ID of node/peer taking this position.
*/
ConsensusProposal(
LedgerId const& prevLedger,
std::uint32_t seq,
Position const& position,
NetClock::time_point closeTime,
NetClock::time_point now,
NodeId const& nodeID)
: previousLedger_(prevLedger)
, position_(position)
, closeTime_(closeTime)
, time_(now)
, proposeSeq_(seq)
, nodeID_(nodeID)
{
}
/**
* Identifying which peer took this position.
*/
NodeId const&
nodeID() const
{
return nodeID_;
}
/**
* Get the proposed position.
*/
Position const&
position() const
{
return position_;
}
/**
* Get the prior accepted ledger this position is based on.
*/
LedgerId const&
prevLedger() const
{
return previousLedger_;
}
/**
* Get the sequence number of this proposal
*
* Starting with an initial sequence number of `kSeqJoin`, successive
* proposals from a peer will increase the sequence number.
*
* @return the sequence number
*/
std::uint32_t
proposeSeq() const
{
return proposeSeq_;
}
/**
* The current position on the consensus close time.
*/
NetClock::time_point const&
closeTime() const
{
return closeTime_;
}
/**
* Get when this position was taken.
*/
NetClock::time_point const&
seenTime() const
{
return time_;
}
/**
* Whether this is the first position taken during the current
* consensus round.
*/
bool
isInitial() const
{
return proposeSeq_ == kSeqJoin;
}
/**
* Get whether this node left the consensus process
*/
bool
isBowOut() const
{
return proposeSeq_ == kSeqLeave;
}
/**
* Get whether this position is stale relative to the provided cutoff
*/
bool
isStale(NetClock::time_point cutoff) const
{
return time_ <= cutoff;
}
/**
* Update the position during the consensus process. This will increment
* the proposal's sequence number if it has not already bowed out.
*
* @param newPosition The new position taken.
* @param newCloseTime The new close time.
* @param now the time The new position was taken
*/
void
changePosition(
Position const& newPosition,
NetClock::time_point newCloseTime,
NetClock::time_point now)
{
signingHash_.reset();
position_ = newPosition;
closeTime_ = newCloseTime;
time_ = now;
if (proposeSeq_ != kSeqLeave)
++proposeSeq_;
}
/**
* Leave consensus
*
* Update position to indicate the node left consensus.
*
* @param now Time when this node left consensus.
*/
void
bowOut(NetClock::time_point now)
{
signingHash_.reset();
time_ = now;
proposeSeq_ = kSeqLeave;
}
std::string
render() const
{
std::stringstream ss;
ss << "proposal: previous_ledger: " << previousLedger_ << " proposal_seq: " << proposeSeq_
<< " position: " << position_ << " close_time: " << to_string(closeTime_)
<< " now: " << to_string(time_) << " is_bow_out:" << isBowOut()
<< " node_id: " << nodeID_;
return ss.str();
}
/**
* Get JSON representation for debugging
*/
json::Value
getJson() const
{
using std::to_string;
json::Value ret = json::ValueType::Object;
ret[jss::previous_ledger] = to_string(prevLedger());
if (!isBowOut())
{
ret[jss::transaction_hash] = to_string(position());
ret[jss::propose_seq] = proposeSeq();
}
ret[jss::close_time] = to_string(closeTime().time_since_epoch().count());
return ret;
}
/**
* The digest for this proposal, used for signing purposes.
*/
uint256 const&
signingHash() const
{
if (!signingHash_)
{
signingHash_ = sha512Half(
HashPrefix::Proposal,
std::uint32_t(proposeSeq()),
closeTime().time_since_epoch().count(),
prevLedger(),
position());
}
return signingHash_.value();
}
private:
/**
* Unique identifier of prior ledger this proposal is based on
*/
LedgerId previousLedger_;
/**
* Unique identifier of the position this proposal is taking
*/
Position position_;
/**
* The ledger close time this position is taking
*/
NetClock::time_point closeTime_;
// !The time this position was last updated
NetClock::time_point time_;
/**
* The sequence number of these positions taken by this node
*/
std::uint32_t proposeSeq_;
/**
* The identifier of the node taking this position
*/
NodeId nodeID_;
/**
* The signing hash for this proposal
*/
mutable std::optional<uint256> signingHash_;
};
template <class NodeId, class LedgerId, class Position>
bool
operator==(
ConsensusProposal<NodeId, LedgerId, Position> const& a,
ConsensusProposal<NodeId, LedgerId, Position> const& b)
{
return a.nodeID() == b.nodeID() && a.proposeSeq() == b.proposeSeq() &&
a.prevLedger() == b.prevLedger() && a.position() == b.position() &&
a.closeTime() == b.closeTime() && a.seenTime() == b.seenTime();
}
} // namespace xrpl

View File

@@ -0,0 +1,255 @@
#pragma once
#include <xrpl/basics/UnorderedContainers.h>
#include <xrpl/basics/chrono.h>
#include <xrpl/beast/utility/instrumentation.h>
#include <xrpl/consensus/ConsensusProposal.h>
#include <xrpl/consensus/DisputedTx.h>
#include <chrono>
#include <cstddef>
#include <map>
#include <string>
namespace xrpl {
/**
* Represents how a node currently participates in Consensus.
*
* A node participates in consensus in varying modes, depending on how
* the node was configured by its operator and how well it stays in sync
* with the network during consensus.
*
* @code
* proposing observing
* \ /
* \---> wrongLedger <---/
* ^
* |
* |
* v
* switchedLedger
* @endcode
*
* We enter the round proposing or observing. If we detect we are working
* on the wrong prior ledger, we go to wrongLedger and attempt to acquire
* the right one. Once we acquire the right one, we go to the switchedLedger
* mode. It is possible we fall behind again and find there is a new better
* ledger, moving back and forth between wrongLedger and switchLedger as
* we attempt to catch up.
*/
enum class ConsensusMode {
/**
* We are normal participant in consensus and propose our position
*/
Proposing,
/**
* We are observing peer positions, but not proposing our position
*/
Observing,
/**
* We have the wrong ledger and are attempting to acquire it
*/
WrongLedger,
/**
* We switched ledgers since we started this consensus round but are now
* running on what we believe is the correct ledger. This mode is as
* if we entered the round observing, but is used to indicate we did
* have the wrongLedger at some point.
*/
SwitchedLedger
};
inline std::string
to_string(ConsensusMode m)
{
switch (m)
{
case ConsensusMode::Proposing:
return "proposing";
case ConsensusMode::Observing:
return "observing";
case ConsensusMode::WrongLedger:
return "wrongLedger";
case ConsensusMode::SwitchedLedger:
return "switchedLedger";
default:
return "unknown";
}
}
/**
* Phases of consensus for a single ledger round.
*
* @code
* "close" "accept"
* open ------- > establish ---------> accepted
* ^ | |
* |---------------| |
* ^ "startRound" |
* |------------------------------------|
* @endcode
*
* The typical transition goes from open to establish to accepted and
* then a call to startRound begins the process anew. However, if a wrong prior
* ledger is detected and recovered during the establish or accept phase,
* consensus will internally go back to open (see Consensus::handleWrongLedger).
*/
enum class ConsensusPhase {
/**
* We haven't closed our ledger yet, but others might have
*/
Open,
/**
* Establishing consensus by exchanging proposals with our peers
*/
Establish,
/**
* We have accepted a new last closed ledger and are waiting on a call
* to startRound to begin the next consensus round. No changes
* to consensus phase occur while in this phase.
*/
Accepted,
};
inline std::string
to_string(ConsensusPhase p)
{
switch (p)
{
case ConsensusPhase::Open:
return "open";
case ConsensusPhase::Establish:
return "establish";
case ConsensusPhase::Accepted:
return "accepted";
default:
return "unknown";
}
}
/**
* Measures the duration of phases of consensus
*/
class ConsensusTimer
{
using time_point = std::chrono::steady_clock::time_point;
time_point start_;
std::chrono::milliseconds dur_{};
public:
[[nodiscard]] std::chrono::milliseconds
read() const
{
return dur_;
}
void
tick(std::chrono::milliseconds fixed)
{
dur_ += fixed;
}
void
reset(time_point tp)
{
start_ = tp;
dur_ = std::chrono::milliseconds{0};
}
void
tick(time_point tp)
{
using namespace std::chrono;
dur_ = duration_cast<milliseconds>(tp - start_);
}
};
/**
* Stores the set of initial close times
*
* The initial consensus proposal from each peer has that peer's view of
* when the ledger closed. This object stores all those close times for
* analysis of clock drift between peers.
*/
struct ConsensusCloseTimes
{
explicit ConsensusCloseTimes() = default;
/**
* Close time estimates, keep ordered for predictable traverse
*/
std::map<NetClock::time_point, int> peers;
/**
* Our close time estimate
*/
NetClock::time_point self;
};
/**
* Whether we have or don't have a consensus
*/
enum class ConsensusState {
No, ///< We do not have consensus
MovedOn, ///< The network has consensus without us
Expired, ///< Consensus time limit has hard-expired
Yes ///< We have consensus along with the network
};
/**
* Encapsulates the result of consensus.
*
* Stores all relevant data for the outcome of consensus on a single
* ledger.
*
* @tparam Traits Traits class defining the concrete consensus types used
* by the application.
*/
template <class Traits>
struct ConsensusResult
{
using Ledger_t = Traits::Ledger_t;
using TxSet_t = Traits::TxSet_t;
using NodeID_t = Traits::NodeID_t;
using Tx_t = TxSet_t::Tx;
using Proposal_t = ConsensusProposal<NodeID_t, typename Ledger_t::ID, typename TxSet_t::ID>;
using Dispute_t = DisputedTx<Tx_t, NodeID_t>;
ConsensusResult(TxSet_t&& s, Proposal_t&& p) : txns{std::move(s)}, position{std::move(p)}
{
XRPL_ASSERT(txns.id() == position.position(), "xrpl::ConsensusResult : valid inputs");
}
/**
* The set of transactions consensus agrees go in the ledger
*/
TxSet_t txns;
/**
* Our proposed position on transactions/close time
*/
Proposal_t position;
/**
* Transactions which are under dispute with our peers
*/
hash_map<typename Tx_t::ID, Dispute_t> disputes;
// Set of TxSet ids we have already compared/created disputes
hash_set<typename TxSet_t::ID> compares;
// Measures the duration of the establish phase for this consensus round
ConsensusTimer roundTime;
// Indicates state in which consensus ended. Once in the accept phase
// will be either Yes or MovedOn or Expired
ConsensusState state = ConsensusState::No;
// The number of peers proposing during the round
std::size_t proposers = 0;
};
} // namespace xrpl

View File

@@ -0,0 +1,367 @@
#pragma once
#include <xrpl/basics/Log.h>
#include <xrpl/beast/utility/Journal.h>
#include <xrpl/consensus/ConsensusParms.h>
#include <xrpl/json/json_value.h>
#include <xrpl/json/json_writer.h>
#include <boost/container/flat_map.hpp>
#include <cstddef>
#include <memory>
#include <sstream>
#include <string>
#include <utility>
namespace xrpl {
/**
* A transaction discovered to be in dispute during consensus.
*
* During consensus, a @ref DisputedTx is created when a transaction
* is discovered to be disputed. The object persists only as long as
* the dispute.
*
* Undisputed transactions have no corresponding @ref DisputedTx object.
*
* Refer to @ref Consensus for details on the template type requirements.
*
* @tparam Tx The type for a transaction
* @tparam NodeId The type for a node identifier
*/
template <class Tx, class NodeId>
class DisputedTx
{
using TxID_t = Tx::ID;
using Map_t = boost::container::flat_map<NodeId, bool>;
public:
/**
* Constructor
*
* @param tx The transaction under dispute
* @param ourVote Our vote on whether tx should be included
* @param numPeers Anticipated number of peer votes
* @param j Journal for debugging
*/
DisputedTx(Tx tx, bool ourVote, std::size_t numPeers, beast::Journal j)
: ourVote_(ourVote), tx_(std::move(tx)), j_(j)
{
votes_.reserve(numPeers);
}
/**
* The unique id/hash of the disputed transaction.
*/
[[nodiscard]] TxID_t const&
id() const
{
return tx_.id();
}
/**
* Our vote on whether the transaction should be included.
*/
[[nodiscard]] bool
getOurVote() const
{
return ourVote_;
}
/**
* Are we and our peers "stalled" where we probably won't change
* our vote?
*/
[[nodiscard]] bool
stalled(
ConsensusParms const& p,
bool proposing,
int peersUnchanged,
beast::Journal j,
std::unique_ptr<std::stringstream> const& clog) const
{
// at() can throw, but the map is built by hand to ensure all valid
// values are available.
auto const& currentCutoff = p.avalancheCutoffs.at(avalancheState_);
auto const& nextCutoff = p.avalancheCutoffs.at(currentCutoff.next);
// We're have not reached the final avalanche state, or been there long
// enough, so there's room for change. Check the times in case the state
// machine is altered to allow states to loop.
if (nextCutoff.consensusTime > currentCutoff.consensusTime ||
avalancheCounter_ < p.avMinRounds)
return false;
// We've haven't had this vote for minimum rounds yet. Things could
// change.
if (proposing && currentVoteCounter_ < p.avMinRounds)
return false;
// If we or any peers have changed a vote in several rounds, then
// things could still change. But if _either_ has not changed in that
// long, we're unlikely to change our vote any time soon. (This prevents
// a malicious peer from flip-flopping a vote to prevent consensus.)
if (peersUnchanged < p.avStalledRounds &&
(proposing && currentVoteCounter_ < p.avStalledRounds))
return false;
// Does this transaction have more than 80% agreement
// Compute the percentage of nodes voting 'yes' (possibly including us)
int const support = (yays_ + (proposing && ourVote_ ? 1 : 0)) * 100;
int const total = nays_ + yays_ + (proposing ? 1 : 0);
if (total == 0)
{
// There are no votes, so we know nothing
return false;
}
int const weight = support / total;
// Returns true if the tx has more than minCONSENSUS_PCT (80) percent
// agreement. Either voting for _or_ voting against the tx.
bool const stalled = weight > p.minConsensusPct || weight < (100 - p.minConsensusPct);
if (stalled)
{
// stalling is an error condition for even a single
// transaction.
std::stringstream s;
s << "Transaction " << id() << " is stalled. We have been voting "
<< (getOurVote() ? "YES" : "NO") << " for " << currentVoteCounter_
<< " rounds. Peers have not changed their votes in " << peersUnchanged
<< " rounds. The transaction has " << weight << "% support. ";
JLOG(j_.error()) << s.str();
CLOG(clog) << s.str();
}
return stalled;
}
/**
* The disputed transaction.
*/
[[nodiscard]] Tx const&
tx() const
{
return tx_;
}
/**
* Change our vote
*/
void
setOurVote(bool o)
{
ourVote_ = o;
}
/**
* Change a peer's vote
*
* @param peer Identifier of peer.
* @param votesYes Whether peer votes to include the disputed transaction.
*
* @return bool Whether the peer changed its vote. (A new vote counts as a
* change.)
*/
[[nodiscard]] bool
setVote(NodeId const& peer, bool votesYes);
/**
* Remove a peer's vote
*
* @param peer Identifier of peer.
*/
void
unVote(NodeId const& peer);
/**
* Update our vote given progression of consensus.
*
* Updates our vote on this disputed transaction based on our peers' votes
* and how far along consensus has proceeded.
*
* @param percentTime Percentage progress through consensus, e.g. 50%
* through or 90%.
* @param proposing Whether we are proposing to our peers in this round.
* @param p Consensus parameters controlling thresholds for voting
* @return Whether our vote changed
*/
bool
updateVote(int percentTime, bool proposing, ConsensusParms const& p);
/**
* JSON representation of dispute, used for debugging
*/
[[nodiscard]] json::Value
getJson() const;
private:
int yays_{0}; //< Number of yes votes
int nays_{0}; //< Number of no votes
bool ourVote_; //< Our vote (true is yes)
Tx tx_; //< Transaction under dispute
Map_t votes_; //< Map from NodeID to vote
/**
* The number of rounds we've gone without changing our vote
*/
std::size_t currentVoteCounter_ = 0;
/**
* Which minimum acceptance percentage phase we are currently in
*/
ConsensusParms::AvalancheState avalancheState_ = ConsensusParms::AvalancheState::Init;
/**
* How long we have been in the current acceptance phase
*/
std::size_t avalancheCounter_ = 0;
beast::Journal const j_;
};
// Track a peer's yes/no vote on a particular disputed tx_
template <class Tx, class NodeId>
bool
DisputedTx<Tx, NodeId>::setVote(NodeId const& peer, bool votesYes)
{
auto const [it, inserted] = votes_.insert(std::make_pair(peer, votesYes));
// new vote
if (inserted)
{
if (votesYes)
{
JLOG(j_.debug()) << "Peer " << peer << " votes YES on " << tx_.id();
++yays_;
}
else
{
JLOG(j_.debug()) << "Peer " << peer << " votes NO on " << tx_.id();
++nays_;
}
return true;
}
// changes vote to yes
if (votesYes && !it->second)
{
JLOG(j_.debug()) << "Peer " << peer << " now votes YES on " << tx_.id();
--nays_;
++yays_;
it->second = true;
return true;
}
// changes vote to no
if (!votesYes && it->second)
{
JLOG(j_.debug()) << "Peer " << peer << " now votes NO on " << tx_.id();
++nays_;
--yays_;
it->second = false;
return true;
}
return false;
}
// Remove a peer's vote on this disputed transaction
template <class Tx, class NodeId>
void
DisputedTx<Tx, NodeId>::unVote(NodeId const& peer)
{
auto it = votes_.find(peer);
if (it != votes_.end())
{
if (it->second)
{
--yays_;
}
else
{
--nays_;
}
votes_.erase(it);
}
}
template <class Tx, class NodeId>
bool
DisputedTx<Tx, NodeId>::updateVote(int percentTime, bool proposing, ConsensusParms const& p)
{
if (ourVote_ && (nays_ == 0))
return false;
if (!ourVote_ && (yays_ == 0))
return false;
bool newPosition = false;
int weight = 0;
// When proposing, to prevent avalanche stalls, we increase the needed
// weight slightly over time. We also need to ensure that the consensus has
// made a minimum number of attempts at each "state" before moving
// to the next.
// Proposing or not, we need to keep track of which state we've reached so
// we can determine if the vote has stalled.
auto const [requiredPct, newState] =
getNeededWeight(p, avalancheState_, percentTime, ++avalancheCounter_, p.avMinRounds);
if (newState)
{
avalancheState_ = *newState;
avalancheCounter_ = 0;
}
if (proposing) // give ourselves full weight
{
// This is basically the percentage of nodes voting 'yes' (including us)
weight = ((yays_ * 100) + (ourVote_ ? 100 : 0)) / (nays_ + yays_ + 1);
newPosition = weight > requiredPct;
}
else
{
// don't let us outweigh a proposing node, just recognize consensus
weight = -1;
newPosition = yays_ > nays_;
}
if (newPosition == ourVote_)
{
++currentVoteCounter_;
JLOG(j_.info()) << "No change (" << (ourVote_ ? "YES" : "NO") << ") on " << tx_.id()
<< " : weight " << weight << ", percent " << percentTime
<< ", round(s) with this vote: " << currentVoteCounter_;
JLOG(j_.debug()) << json::Compact{getJson()};
return false;
}
currentVoteCounter_ = 0;
ourVote_ = newPosition;
JLOG(j_.debug()) << "We now vote " << (ourVote_ ? "YES" : "NO") << " on " << tx_.id();
JLOG(j_.debug()) << json::Compact{getJson()};
return true;
}
template <class Tx, class NodeId>
json::Value
DisputedTx<Tx, NodeId>::getJson() const
{
using std::to_string;
json::Value ret(json::ValueType::Object);
ret["yays"] = yays_;
ret["nays"] = nays_;
ret["our_vote"] = ourVote_;
if (!votes_.empty())
{
json::Value votes(json::ValueType::Object);
for (auto const& [nodeId, vote] : votes_)
votes[to_string(nodeId)] = vote;
ret["votes"] = std::move(votes);
}
return ret;
}
} // namespace xrpl

View File

@@ -0,0 +1,846 @@
#pragma once
#include <xrpl/basics/ToString.h>
#include <xrpl/beast/utility/instrumentation.h>
#include <xrpl/json/json_value.h>
#include <algorithm>
#include <cstddef>
#include <cstdint>
#include <iomanip>
#include <map>
#include <memory>
#include <optional>
#include <ostream>
#include <sstream>
#include <stack>
#include <utility>
#include <vector>
namespace xrpl {
/**
* The tip of a span of ledger ancestry
*/
template <class Ledger>
class SpanTip
{
public:
using Seq = Ledger::Seq;
using ID = Ledger::ID;
SpanTip(Seq s, ID i, Ledger const lgr) : seq{s}, id{i}, ledger_{std::move(lgr)}
{
}
// The sequence number of the tip ledger
Seq seq;
// The ID of the tip ledger
ID id;
/**
* Lookup the ID of an ancestor of the tip ledger
*
* @param s The sequence number of the ancestor
* @return The ID of the ancestor with that sequence number
*
* @note s must be less than or equal to the sequence number of the
* tip ledger
*/
[[nodiscard]] ID
ancestor(Seq const& s) const
{
XRPL_ASSERT(s <= seq, "xrpl::SpanTip::ancestor : valid input");
return ledger_[s];
}
private:
Ledger const ledger_;
};
namespace ledger_trie_detail {
// Represents a span of ancestry of a ledger
template <class Ledger>
class Span
{
using Seq = Ledger::Seq;
using ID = Ledger::ID;
// The span is the half-open interval [start,end) of ledger_
Seq start_{0};
Seq end_{1};
Ledger ledger_;
public:
Span() : ledger_{typename Ledger::MakeGenesis{}}
{
// Require default ledger to be genesis seq
XRPL_ASSERT(ledger_.seq() == start_, "xrpl::Span::Span : ledger is genesis");
}
Span(Ledger ledger) : end_{ledger.seq() + Seq{1}}, ledger_{std::move(ledger)}
{
}
Span(Span const& s) = default;
Span(Span&& s) = default;
Span&
operator=(Span const&) = default;
Span&
operator=(Span&&) = default;
[[nodiscard]] Seq
start() const
{
return start_;
}
[[nodiscard]] Seq
end() const
{
return end_;
}
// Return the Span from [spot,end_) or none if no such valid span
[[nodiscard]] std::optional<Span>
from(Seq spot) const
{
return sub(spot, end_);
}
// Return the Span from [start_,spot) or none if no such valid span
[[nodiscard]] std::optional<Span>
before(Seq spot) const
{
return sub(start_, spot);
}
// Return the ID of the ledger that starts this span
[[nodiscard]] ID
startID() const
{
return ledger_[start_];
}
// Return the ledger sequence number of the first possible difference
// between this span and a given ledger.
[[nodiscard]] Seq
diff(Ledger const& o) const
{
return clamp(mismatch(ledger_, o));
}
// The tip of this span
[[nodiscard]] SpanTip<Ledger>
tip() const
{
Seq const tipSeq{end_ - Seq{1}};
return SpanTip<Ledger>{tipSeq, ledger_[tipSeq], ledger_};
}
private:
Span(Seq start, Seq end, Ledger l) : start_{start}, end_{end}, ledger_{std::move(l)}
{
// Spans cannot be empty
XRPL_ASSERT(start < end, "xrpl::Span::Span : non-empty span input");
}
[[nodiscard]] Seq
clamp(Seq val) const
{
return std::min(std::max(start_, val), end_);
}
// Return a span of this over the half-open interval [from,to)
[[nodiscard]] std::optional<Span>
sub(Seq from, Seq to) const
{
Seq const newFrom = clamp(from);
Seq const newTo = clamp(to);
if (newFrom < newTo)
return Span(newFrom, newTo, ledger_);
return std::nullopt;
}
friend std::ostream&
operator<<(std::ostream& o, Span const& s)
{
return o << s.tip().id << "[" << s.start_ << "," << s.end_ << ")";
}
friend Span
merge(Span const& a, Span const& b)
{
// Return combined span, using ledger_ from higher sequence span
if (a.end_ < b.end_)
return Span(std::min(a.start_, b.start_), b.end_, b.ledger_);
return Span(std::min(a.start_, b.start_), a.end_, a.ledger_);
}
};
// A node in the trie
template <class Ledger>
struct Node
{
Node() = default;
explicit Node(Ledger const& l) : span{l}, tipSupport{1}, branchSupport{1}
{
}
explicit Node(Span<Ledger> s) : span{std::move(s)}
{
}
Span<Ledger> span;
std::uint32_t tipSupport = 0;
std::uint32_t branchSupport = 0;
std::vector<std::unique_ptr<Node>> children;
Node* parent = nullptr;
/**
* Remove the given node from this Node's children
*
* @param child The address of the child node to remove
* @note The child must be a member of the vector. The passed pointer
* will be dangling as a result of this call
*/
void
erase(Node const* child)
{
auto it = std::ranges::find_if(
children, [child](std::unique_ptr<Node> const& curr) { return curr.get() == child; });
XRPL_ASSERT(it != children.end(), "xrpl::Node::erase : valid input");
std::swap(*it, children.back());
children.pop_back();
}
friend std::ostream&
operator<<(std::ostream& o, Node const& s)
{
return o << s.span << "(T:" << s.tipSupport << ",B:" << s.branchSupport << ")";
}
[[nodiscard]] json::Value
getJson() const
{
json::Value res;
std::stringstream sps;
sps << span;
res["span"] = sps.str();
res["startID"] = to_string(span.startID());
res["seq"] = static_cast<std::uint32_t>(span.tip().seq);
res["tipSupport"] = tipSupport;
res["branchSupport"] = branchSupport;
if (!children.empty())
{
json::Value& cs = (res["children"] = json::ValueType::Array);
for (auto const& child : children)
{
cs.append(child->getJson());
}
}
return res;
}
};
} // namespace ledger_trie_detail
/**
* Ancestry trie of ledgers
*
* A compressed trie tree that maintains validation support of recent ledgers
* based on their ancestry.
*
* The compressed trie structure comes from recognizing that ledger history
* can be viewed as a string over the alphabet of ledger ids. That is,
* a given ledger with sequence number `seq` defines a length `seq` string,
* with i-th entry equal to the id of the ancestor ledger with sequence
* number i. "Sequence" strings with a common prefix share those ancestor
* ledgers in common. Tracking this ancestry information and relations across
* all validated ledgers is done conveniently in a compressed trie. A node in
* the trie is an ancestor of all its children. If a parent node has sequence
* number `seq`, each child node has a different ledger starting at `seq+1`.
* The compression comes from the invariant that any non-root node with 0 tip
* support has either no children or multiple children. In other words, a
* non-root 0-tip-support node can be combined with its single child.
*
* Each node has a tipSupport, which is the number of current validations for
* that particular ledger. The node's branch support is the sum of the tip
* support and the branch support of that node's children:
*
* @code
* node->branchSupport = node->tipSupport;
* for (child : node->children)
* node->branchSupport += child->branchSupport;
* @endcode
*
* The templated Ledger type represents a ledger which has a unique history.
* It should be lightweight and cheap to copy.
*
* @code
* // Identifier types that should be equality-comparable and copyable
* struct ID;
* struct Seq;
*
* struct Ledger
* {
* struct MakeGenesis{};
*
* // The genesis ledger represents a ledger that prefixes all other
* // ledgers
* Ledger(MakeGenesis{});
*
* Ledger(Ledger const&);
* Ledger& operator=(Ledger const&);
*
* // Return the sequence number of this ledger
* Seq seq() const;
*
* // Return the ID of this ledger's ancestor with given sequence number
* // or ID{0} if unknown
* ID
* operator[](Seq s);
*
* };
*
* // Return the sequence number of the first possible mismatching ancestor
* // between two ledgers
* Seq
* mismatch(ledgerA, ledgerB);
* @endcode
*
* The unique history invariant of ledgers requires any ledgers that agree
* on the id of a given sequence number agree on ALL ancestors before that
* ledger:
*
* @code
* Ledger a,b;
* // For all Seq s:
* if(a[s] == b[s]);
* for(Seq p = 0; p < s; ++p)
* assert(a[p] == b[p]);
* @endcode
*
* @tparam Ledger A type representing a ledger and its history
*/
template <class Ledger>
class LedgerTrie
{
using Seq = Ledger::Seq;
using ID = Ledger::ID;
using Node = ledger_trie_detail::Node<Ledger>;
using Span = ledger_trie_detail::Span<Ledger>;
// The root of the trie. The root is allowed to break the no-single child
// invariant.
std::unique_ptr<Node> root_;
// Count of the tip support for each sequence number
std::map<Seq, std::uint32_t> seqSupport_;
/**
* Find the node in the trie that represents the longest common ancestry
* with the given ledger.
*
* @return Pair of the found node and the sequence number of the first
* ledger difference.
*/
[[nodiscard]] std::pair<Node*, Seq>
find(Ledger const& ledger) const
{
// NOLINTNEXTLINE(misc-const-correctness)
Node* curr = root_.get();
// Root is always defined and is in common with all ledgers
XRPL_ASSERT(curr, "xrpl::LedgerTrie::find : non-null root");
Seq pos = curr->span.diff(ledger);
bool done = false;
// Continue searching for a better span as long as the current position
// matches the entire span
while (!done && pos == curr->span.end())
{
done = true;
// Find the child with the longest ancestry match
for (std::unique_ptr<Node> const& child : curr->children)
{
auto const childPos = child->span.diff(ledger);
if (childPos > pos)
{
done = false;
pos = childPos;
curr = child.get();
break;
}
}
}
return std::make_pair(curr, pos);
}
/**
* Find the node in the trie with an exact match to the given ledger ID
*
* @return the found node or nullptr if an exact match was not found.
*
* @note O(n) since this searches all nodes until a match is found
*/
Node*
findByLedgerID(Ledger const& ledger, Node* parent = nullptr) const
{
if (parent == nullptr)
parent = root_.get();
if (ledger.id() == parent->span.tip().id)
return parent;
for (auto const& child : parent->children)
{
auto cl = findByLedgerID(ledger, child.get());
if (cl)
return cl;
}
return nullptr;
}
void
dumpImpl(std::ostream& o, std::unique_ptr<Node> const& curr, int offset) const
{
if (curr)
{
if (offset > 0)
o << std::setw(offset) << "|-";
std::stringstream ss;
ss << *curr;
o << ss.str() << std::endl;
for (std::unique_ptr<Node> const& child : curr->children)
dumpImpl(o, child, offset + 1 + ss.str().size() + 2);
}
}
public:
LedgerTrie() : root_{std::make_unique<Node>()}
{
}
/**
* Insert and/or increment the support for the given ledger.
*
* @param ledger A ledger and its ancestry
* @param count The count of support for this ledger
*/
void
insert(Ledger const& ledger, std::uint32_t count = 1)
{
auto const [loc, diffSeq] = find(ledger);
// There is always a place to insert
XRPL_ASSERT(loc, "xrpl::LedgerTrie::insert : valid input ledger");
// Node from which to start incrementing branchSupport
Node* incNode = loc;
// loc->span has the longest common prefix with Span{ledger} of all
// existing nodes in the trie. The optional<Span>'s below represent
// the possible common suffixes between loc->span and Span{ledger}.
//
// loc->span
// a b c | d e f
// prefix | oldSuffix
//
// Span{ledger}
// a b c | g h i
// prefix | newSuffix
std::optional<Span> prefix = loc->span.before(diffSeq);
std::optional<Span> oldSuffix = loc->span.from(diffSeq);
std::optional<Span> newSuffix = Span{ledger}.from(diffSeq);
if (oldSuffix)
{
// Have
// abcdef -> ....
// Inserting
// abc
// Becomes
// abc -> def -> ...
// Create oldSuffix node that takes over loc
auto newNode = std::make_unique<Node>(*oldSuffix);
newNode->tipSupport = loc->tipSupport;
newNode->branchSupport = loc->branchSupport;
newNode->children = std::move(loc->children);
XRPL_ASSERT(loc->children.empty(), "xrpl::LedgerTrie::insert : moved-from children");
for (std::unique_ptr<Node>& child : newNode->children)
child->parent = newNode.get();
// Loc truncates to prefix and newNode is its child
XRPL_ASSERT(prefix, "xrpl::LedgerTrie::insert : prefix is set");
loc->span = *prefix; // NOLINT(bugprone-unchecked-optional-access) assert above
newNode->parent = loc;
loc->children.emplace_back(std::move(newNode));
loc->tipSupport = 0;
}
if (newSuffix)
{
// Have
// abc -> ...
// Inserting
// abcdef-> ...
// Becomes
// abc -> ...
// \-> def
auto newNode = std::make_unique<Node>(*newSuffix);
newNode->parent = loc;
// increment support starting from the new node
incNode = newNode.get();
loc->children.push_back(std::move(newNode));
}
incNode->tipSupport += count;
while (incNode)
{
incNode->branchSupport += count;
incNode = incNode->parent;
}
seqSupport_[ledger.seq()] += count;
}
/**
* Decrease support for a ledger, removing and compressing if possible.
*
* @param ledger The ledger history to remove
* @param count The amount of tip support to remove
*
* @return Whether a matching node was decremented and possibly removed.
*/
bool
remove(Ledger const& ledger, std::uint32_t count = 1)
{
Node* loc = findByLedgerID(ledger);
// Must be exact match with tip support
if ((loc == nullptr) || loc->tipSupport == 0)
return false;
// found our node, remove it
count = std::min(count, loc->tipSupport);
loc->tipSupport -= count;
auto const it = seqSupport_.find(ledger.seq());
XRPL_ASSERT(
it != seqSupport_.end() && it->second >= count,
"xrpl::LedgerTrie::remove : valid input ledger");
it->second -= count;
if (it->second == 0)
seqSupport_.erase(it->first);
Node* decNode = loc;
while (decNode)
{
decNode->branchSupport -= count;
decNode = decNode->parent;
}
while (loc->tipSupport == 0 && loc != root_.get())
{
Node* parent = loc->parent;
if (loc->children.empty())
{
// this node can be erased
parent->erase(loc);
}
else if (loc->children.size() == 1)
{
// This node can be combined with its child
std::unique_ptr<Node> child = std::move(loc->children.front());
child->span = merge(loc->span, child->span);
child->parent = parent;
parent->children.emplace_back(std::move(child));
parent->erase(loc);
}
else
{
break;
}
loc = parent;
}
return true;
}
/**
* Return count of tip support for the specific ledger.
*
* @param ledger The ledger to lookup
* @return The number of entries in the trie for this *exact* ledger
*/
[[nodiscard]] std::uint32_t
tipSupport(Ledger const& ledger) const
{
if (auto const* loc = findByLedgerID(ledger))
return loc->tipSupport;
return 0;
}
/**
* Return the count of branch support for the specific ledger
*
* @param ledger The ledger to lookup
* @return The number of entries in the trie for this ledger or a
* descendant
*/
[[nodiscard]] std::uint32_t
branchSupport(Ledger const& ledger) const
{
Node const* loc = findByLedgerID(ledger);
if (loc == nullptr)
{
Seq diffSeq;
std::tie(loc, diffSeq) = find(ledger);
// Check that ledger is a proper prefix of loc
if (!(diffSeq > ledger.seq() && ledger.seq() < loc->span.end()))
loc = nullptr;
}
return loc ? loc->branchSupport : 0;
}
/**
* Return the preferred ledger ID
*
* The preferred ledger is used to determine the working ledger
* for consensus amongst competing alternatives.
*
* Recall that each validator is normally validating a chain of ledgers,
* e.g. A->B->C->D. However, if due to network connectivity or other
* issues, validators generate different chains
*
* @code
* /->C
* A->B
* \->D->E
* @endcode
*
* we need a way for validators to converge on the chain with the most
* support. We call this the preferred ledger. Intuitively, the idea is to
* be conservative and only switch to a different branch when you see
* enough peer validations to *know* another branch won't have preferred
* support.
*
* The preferred ledger is found by walking this tree of validated ledgers
* starting from the common ancestor ledger.
*
* At each sequence number, we have
*
* - The prior sequence preferred ledger, e.g. B.
* - The (tip) support of ledgers with this sequence number,e.g. the
* number of validators whose last validation was for C or D.
* - The (branch) total support of all descendants of the current
* sequence number ledgers, e.g. the branch support of D is the
* tip support of D plus the tip support of E; the branch support of
* C is just the tip support of C.
* - The number of validators that have yet to validate a ledger
* with this sequence number (uncommitted support). Uncommitted
* includes all validators whose last sequence number is smaller than
* our last issued sequence number, since due to asynchrony, we may
* not have heard from those nodes yet.
*
* The preferred ledger for this sequence number is then the ledger
* with relative majority of support, where uncommitted support
* can be given to ANY ledger at that sequence number
* (including one not yet known). If no such preferred ledger exists, then
* the prior sequence preferred ledger is the overall preferred ledger.
*
* In this example, for D to be preferred, the number of validators
* supporting it or a descendant must exceed the number of validators
* supporting C _plus_ the current uncommitted support. This is because if
* all uncommitted validators end up validating C, that new support must
* be less than that for D to be preferred.
*
* If a preferred ledger does exist, then we continue with the next
* sequence using that ledger as the root.
*
* @param largestIssued The sequence number of the largest validation
* issued by this node.
* @return Pair with the sequence number and ID of the preferred ledger or
* std::nullopt if no preferred ledger exists
*/
[[nodiscard]] std::optional<SpanTip<Ledger>>
getPreferred(Seq const largestIssued) const
{
if (empty())
return std::nullopt;
Node* curr = root_.get();
bool done = false;
std::uint32_t uncommitted = 0;
auto uncommittedIt = seqSupport_.begin();
while (curr && !done)
{
// Within a single span, the preferred by branch strategy is simply
// to continue along the span as long as the branch support of
// the next ledger exceeds the uncommitted support for that ledger.
{
// Add any initial uncommitted support prior for ledgers
// earlier than nextSeq or earlier than largestIssued
Seq nextSeq = curr->span.start() + Seq{1};
while (uncommittedIt != seqSupport_.end() &&
uncommittedIt->first < std::max(nextSeq, largestIssued))
{
uncommitted += uncommittedIt->second;
uncommittedIt++;
}
// Advance nextSeq along the span
while (nextSeq < curr->span.end() && curr->branchSupport > uncommitted)
{
// Jump to the next seqSupport change
if (uncommittedIt != seqSupport_.end() &&
uncommittedIt->first < curr->span.end())
{
nextSeq = uncommittedIt->first + Seq{1};
uncommitted += uncommittedIt->second;
uncommittedIt++;
}
else
{ // otherwise we jump to the end of the span
nextSeq = curr->span.end();
}
}
// We did not consume the entire span, so we have found the
// preferred ledger
if (nextSeq < curr->span.end())
{
// nextSeq within span guarantees before() is set
// NOLINTNEXTLINE(bugprone-unchecked-optional-access)
return curr->span.before(nextSeq)->tip();
}
}
// We have reached the end of the current span, so we need to
// find the best child
Node* best = nullptr;
std::uint32_t margin = 0;
if (curr->children.size() == 1)
{
best = curr->children[0].get();
margin = best->branchSupport;
}
else if (!curr->children.empty())
{
// Sort placing children with largest branch support in the
// front, breaking ties with the span's starting ID
std::partial_sort(
curr->children.begin(),
curr->children.begin() + 2,
curr->children.end(),
[](std::unique_ptr<Node> const& a, std::unique_ptr<Node> const& b) {
return std::make_tuple(a->branchSupport, a->span.startID()) >
std::make_tuple(b->branchSupport, b->span.startID());
});
best = curr->children[0].get();
margin = curr->children[0]->branchSupport - curr->children[1]->branchSupport;
// If best holds the tie-breaker, gets one larger margin
// since the second best needs additional branchSupport
// to overcome the tie
if (best->span.startID() > curr->children[1]->span.startID())
margin++;
}
// If the best child has margin exceeding the uncommitted support,
// continue from that child, otherwise we are done
if (best && ((margin > uncommitted) || (uncommitted == 0)))
{
curr = best;
}
else
{ // current is the best
done = true;
}
}
return curr->span.tip();
}
/**
* Return whether the trie is tracking any ledgers
*/
[[nodiscard]] bool
empty() const
{
return !root_ || root_->branchSupport == 0;
}
/**
* Dump an ascii representation of the trie to the stream
*/
void
dump(std::ostream& o) const
{
dumpImpl(o, root_, 0);
}
/**
* Dump JSON representation of trie state
*/
[[nodiscard]] json::Value
getJson() const
{
json::Value res;
res["trie"] = root_->getJson();
res["seq_support"] = json::ValueType::Object;
for (auto const& [seq, sup] : seqSupport_)
res["seq_support"][to_string(seq)] = sup;
return res;
}
/**
* Check the compressed trie and support invariants.
*/
[[nodiscard]] bool
checkInvariants() const
{
std::map<Seq, std::uint32_t> expectedSeqSupport;
std::stack<Node const*> nodes;
nodes.push(root_.get());
while (!nodes.empty())
{
Node const* curr = nodes.top();
nodes.pop();
if (curr == nullptr)
continue;
// Node with 0 tip support must have multiple children
// unless it is the root node
if (curr != root_.get() && curr->tipSupport == 0 && curr->children.size() < 2)
return false;
// branchSupport = tipSupport + sum(child->branchSupport)
std::size_t support = curr->tipSupport;
if (curr->tipSupport != 0)
expectedSeqSupport[curr->span.end() - Seq{1}] += curr->tipSupport;
for (auto const& child : curr->children)
{
if (child->parent != curr)
return false;
support += child->branchSupport;
nodes.push(child.get());
}
if (support != curr->branchSupport)
return false;
}
return expectedSeqSupport == seqSupport_;
}
};
} // namespace xrpl

View File

@@ -0,0 +1,8 @@
# Consensus
This directory contains the implementation of a
generic consensus algorithm. The implementation
follows a CRTP design, requiring client code to implement
specific functions and types to use consensus in their
application. The interface is undergoing refactoring and
is not yet finalized.

File diff suppressed because it is too large Load Diff

View File

@@ -15,9 +15,9 @@
namespace xrpl {
// Forward declarations
namespace NodeStore {
namespace node_store {
class Database;
} // namespace NodeStore
} // namespace node_store
namespace Resource {
class Manager;
} // namespace Resource
@@ -167,7 +167,7 @@ public:
getResourceManager() = 0;
// Storage services
virtual NodeStore::Database&
virtual node_store::Database&
getNodeStore() = 0;
virtual SHAMapStore&

View File

@@ -13,7 +13,7 @@
#include <stdexcept>
#include <string>
namespace xrpl::NodeStore {
namespace xrpl::node_store {
/**
* A backend used for the NodeStore.
@@ -163,4 +163,4 @@ public:
fdRequired() const = 0;
};
} // namespace xrpl::NodeStore
} // namespace xrpl::node_store

View File

@@ -25,7 +25,7 @@ namespace xrpl {
class Section;
} // namespace xrpl
namespace xrpl::NodeStore {
namespace xrpl::node_store {
/**
* Persistency layer for NodeObject
@@ -248,7 +248,7 @@ protected:
void
storeStats(std::uint64_t count, std::uint64_t sz)
{
XRPL_ASSERT(count <= sz, "xrpl::NodeStore::Database::storeStats : valid inputs");
XRPL_ASSERT(count <= sz, "xrpl::node_store::Database::storeStats : valid inputs");
storeCount_ += count;
storeSz_ += sz;
}
@@ -308,4 +308,4 @@ private:
threadEntry();
};
} // namespace xrpl::NodeStore
} // namespace xrpl::node_store

View File

@@ -9,7 +9,7 @@
#include <memory>
#include <string>
namespace xrpl::NodeStore {
namespace xrpl::node_store {
/* This class has two key-value store Backend objects for persisting SHAMap
* records. This facilitates online deletion of data. New backends are
@@ -38,7 +38,7 @@ public:
*/
virtual void
rotate(
std::unique_ptr<NodeStore::Backend>&& newBackend,
std::unique_ptr<node_store::Backend>&& newBackend,
std::function<void(std::string const& writableName, std::string const& archiveName)> const&
f) = 0;
@@ -56,4 +56,4 @@ public:
setRotationInFlight(bool inFlight) = 0;
};
} // namespace xrpl::NodeStore
} // namespace xrpl::node_store

View File

@@ -3,7 +3,7 @@
#include <xrpl/nodestore/Scheduler.h>
#include <xrpl/nodestore/Task.h>
namespace xrpl::NodeStore {
namespace xrpl::node_store {
/**
* Simple NodeStore Scheduler that just performs the tasks synchronously.
@@ -21,4 +21,4 @@ public:
onBatchWrite(BatchWriteReport const& report) override;
};
} // namespace xrpl::NodeStore
} // namespace xrpl::node_store

View File

@@ -14,7 +14,7 @@ namespace xrpl {
class Section;
} // namespace xrpl
namespace xrpl::NodeStore {
namespace xrpl::node_store {
/**
* Base class for backend factories.
@@ -70,4 +70,4 @@ public:
}
};
} // namespace xrpl::NodeStore
} // namespace xrpl::node_store

View File

@@ -10,7 +10,7 @@
#include <memory>
#include <string>
namespace xrpl::NodeStore {
namespace xrpl::node_store {
/**
* Singleton for managing NodeStore factories and back ends.
@@ -98,4 +98,4 @@ public:
beast::Journal journal) = 0;
};
} // namespace xrpl::NodeStore
} // namespace xrpl::node_store

View File

@@ -8,7 +8,7 @@
#include <cstdint>
#include <memory>
// VFALCO NOTE Intentionally not in the NodeStore namespace
// VFALCO NOTE Intentionally not in the node_store namespace
namespace xrpl {

View File

@@ -4,7 +4,7 @@
#include <chrono>
namespace xrpl::NodeStore {
namespace xrpl::node_store {
enum class FetchType { Synchronous, Async };
@@ -71,4 +71,4 @@ public:
onBatchWrite(BatchWriteReport const& report) = 0;
};
} // namespace xrpl::NodeStore
} // namespace xrpl::node_store

View File

@@ -1,6 +1,6 @@
#pragma once
namespace xrpl::NodeStore {
namespace xrpl::node_store {
/**
* Derived classes perform scheduled tasks.
@@ -17,4 +17,4 @@ struct Task
performScheduledTask() = 0;
};
} // namespace xrpl::NodeStore
} // namespace xrpl::node_store

View File

@@ -5,7 +5,7 @@
#include <memory>
#include <vector>
namespace xrpl::NodeStore {
namespace xrpl::node_store {
// This is only used to pre-allocate the array for
// batch objects and does not affect the amount written.
@@ -36,4 +36,4 @@ enum class Status {
*/
using Batch = std::vector<std::shared_ptr<NodeObject>>;
} // namespace xrpl::NodeStore
} // namespace xrpl::node_store

View File

@@ -9,7 +9,7 @@
#include <memory>
#include <mutex>
namespace xrpl::NodeStore {
namespace xrpl::node_store {
/**
* Batch-writing assist logic.
@@ -86,4 +86,4 @@ private:
Batch writeSet_;
};
} // namespace xrpl::NodeStore
} // namespace xrpl::node_store

View File

@@ -22,7 +22,7 @@
#include <stdexcept>
#include <utility>
namespace xrpl::NodeStore {
namespace xrpl::node_store {
class DatabaseNodeImp : public Database
{
@@ -68,7 +68,7 @@ public:
XRPL_ASSERT(
backend_,
"xrpl::NodeStore::DatabaseNodeImp::DatabaseNodeImp : non-null "
"xrpl::node_store::DatabaseNodeImp::DatabaseNodeImp : non-null "
"backend");
}
@@ -138,4 +138,4 @@ private:
}
};
} // namespace xrpl::NodeStore
} // namespace xrpl::node_store

View File

@@ -16,7 +16,7 @@
#include <mutex>
#include <string>
namespace xrpl::NodeStore {
namespace xrpl::node_store {
class DatabaseRotatingImp : public DatabaseRotating
{
@@ -41,7 +41,7 @@ public:
void
rotate(
std::unique_ptr<NodeStore::Backend>&& newBackend,
std::unique_ptr<node_store::Backend>&& newBackend,
std::function<void(std::string const& writableName, std::string const& archiveName)> const&
f) override;
@@ -94,4 +94,4 @@ private:
forEach(std::function<void(std::shared_ptr<NodeObject>)> f) override;
};
} // namespace xrpl::NodeStore
} // namespace xrpl::node_store

View File

@@ -4,7 +4,7 @@
#include <memory>
namespace xrpl::NodeStore {
namespace xrpl::node_store {
/**
* Parsed key/value blob into NodeObject components.
@@ -49,4 +49,4 @@ private:
int dataBytes_;
};
} // namespace xrpl::NodeStore
} // namespace xrpl::node_store

View File

@@ -12,7 +12,7 @@
#include <memory>
#include <stdexcept>
namespace xrpl::NodeStore {
namespace xrpl::node_store {
/**
* Convert a NodeObject from in-memory to database format.
@@ -68,7 +68,7 @@ class EncodedBlob
public:
explicit EncodedBlob(std::shared_ptr<NodeObject> const& obj)
: size_([&obj]() {
XRPL_ASSERT(obj, "xrpl::NodeStore::EncodedBlob::EncodedBlob : non-null input");
XRPL_ASSERT(obj, "xrpl::node_store::EncodedBlob::EncodedBlob : non-null input");
if (!obj)
throw std::runtime_error("EncodedBlob: unseated std::shared_ptr used.");
@@ -88,7 +88,7 @@ public:
XRPL_ASSERT(
((ptr_ == payload_.data()) && (size_ <= payload_.size())) ||
((ptr_ != payload_.data()) && (size_ > payload_.size())),
"xrpl::NodeStore::EncodedBlob::~EncodedBlob : valid payload "
"xrpl::node_store::EncodedBlob::~EncodedBlob : valid payload "
"pointer");
if (ptr_ != payload_.data())
@@ -114,4 +114,4 @@ public:
}
};
} // namespace xrpl::NodeStore
} // namespace xrpl::node_store

View File

@@ -13,7 +13,7 @@
#include <string>
#include <vector>
namespace xrpl::NodeStore {
namespace xrpl::node_store {
class ManagerImp : public Manager
{
@@ -57,4 +57,4 @@ public:
beast::Journal journal) override;
};
} // namespace xrpl::NodeStore
} // namespace xrpl::node_store

View File

@@ -21,7 +21,7 @@
#include <cstring>
#include <string>
namespace xrpl::NodeStore {
namespace xrpl::node_store {
template <class BufferFactory>
std::pair<void const*, std::size_t>
@@ -269,7 +269,7 @@ nodeobjectCompress(void const* in, std::size_t inSize, BufferFactory&& bf)
case 1: // lz4
{
std::uint8_t* p = nullptr;
auto const lzr = NodeStore::lz4Compress(in, inSize, [&p, &vn, &bf](std::size_t n) {
auto const lzr = node_store::lz4Compress(in, inSize, [&p, &vn, &bf](std::size_t n) {
p = reinterpret_cast<std::uint8_t*>(bf(vn + n));
return p + vn;
});
@@ -316,4 +316,4 @@ filterInner(void* in, std::size_t inSize)
}
}
} // namespace xrpl::NodeStore
} // namespace xrpl::node_store

View File

@@ -6,7 +6,7 @@
#include <cstdint>
#include <type_traits>
namespace xrpl::NodeStore {
namespace xrpl::node_store {
// This is a variant of the base128 varint format from
// google protocol buffers:
@@ -123,4 +123,4 @@ write(nudb::detail::ostream& os, std::size_t t)
writeVarint(os.data(sizeVarint(t)), t);
}
} // namespace xrpl::NodeStore
} // namespace xrpl::node_store

View File

@@ -0,0 +1,163 @@
#pragma once
#include <xrpl/beast/utility/PropertyStream.h>
#include <xrpl/peerfinder/detail/Tuning.h>
#include <cstddef>
#include <cstdint>
#include <optional>
#include <string>
#include <string_view>
namespace xrpl::PeerFinder {
struct PeerLimitConfig
{
std::optional<std::size_t> maxPeers;
std::optional<std::size_t> inPeers;
std::optional<std::size_t> outPeers;
};
/**
* PeerFinder configuration settings.
*/
struct Config
{
/**
* The largest number of public peer slots to allow.
* This includes both inbound and outbound, but does not include
* fixed peers.
*/
std::size_t maxPeers{Tuning::kDefaultMaxPeers};
/**
* The number of automatic outbound connections to maintain.
* Outbound connections are only maintained if autoConnect
* is `true`.
*/
std::size_t outPeers = calcOutPeers(); // Note: relies on `maxPeers` being initialized
/**
* The number of automatic inbound connections to maintain.
* Inbound connections are only maintained if wantIncoming
* is `true`.
*/
std::size_t inPeers{0};
/**
* `true` if we want our IP address kept private.
*/
bool peerPrivate = true;
/**
* `true` if we want to accept incoming connections.
*/
bool wantIncoming{true};
/**
* `true` if we want to establish connections automatically
*/
bool autoConnect{true};
/**
* The listening port number.
*/
std::uint16_t listeningPort{0};
/**
* The set of features we advertise.
*/
std::string features;
/**
* Limit how many incoming connections we allow per IP
*/
int ipLimit{0};
/**
* `true` if we want to verify endpoints in TMEndpoints messages
*/
bool verifyEndpoints = true;
//--------------------------------------------------------------------------
/**
* Returns a suitable value for outPeers according to the rules.
*/
[[nodiscard]] std::size_t
calcOutPeers() const;
/**
* Adjusts the values so they follow the business rules.
*/
void
applyTuning();
/**
* Write the configuration into a property stream
*/
void
onWrite(beast::PropertyStream::Map& map) const;
/**
* Make PeerFinder::Config from peer limit and server mode parameters.
*/
static Config
makeConfig(
bool peerPrivate,
bool standalone,
PeerLimitConfig const& limits,
std::uint16_t port,
bool validationPublicKey,
int ipLimit,
bool verifyEndpoints);
/**
* Compares two configurations for equality field by field.
*/
friend bool
operator==(Config const& lhs, Config const& rhs) = default;
};
//------------------------------------------------------------------------------
/**
* Possible results from activating a slot.
*/
enum class Result { InboundDisabled, DuplicatePeer, IpLimitExceeded, Full, Success };
/**
* @brief Converts a `Result` enum value to its string representation.
*
* This function provides a human-readable string for a given `Result` enum,
* which is useful for logging, debugging, or displaying status messages.
*
* @param result The `Result` enum value to convert.
* @return A `std::string_view` representing the enum value. Returns "unknown"
* if the enum value is not explicitly handled.
*
* @note This function returns a `std::string_view` for performance.
* A `std::string` would need to allocate memory on the heap and copy the
* string literal into it every time the function is called.
*/
inline std::string_view
to_string(Result result) noexcept
{
switch (result)
{
case Result::InboundDisabled:
return "inbound disabled";
case Result::DuplicatePeer:
return "peer already connected";
case Result::IpLimitExceeded:
return "ip limit exceeded";
case Result::Full:
return "slots full";
case Result::Success:
return "success";
}
return "unknown";
}
} // namespace xrpl::PeerFinder

View File

@@ -0,0 +1,179 @@
#pragma once
#include <xrpl/beast/net/IPEndpoint.h>
#include <xrpl/beast/utility/PropertyStream.h>
#include <xrpl/peerfinder/Config.h>
#include <xrpl/peerfinder/Slot.h>
#include <xrpl/peerfinder/Types.h>
#include <xrpl/protocol/PublicKey.h>
#include <boost/asio/ip/tcp.hpp>
#include <memory>
#include <string>
#include <string_view>
#include <utility>
#include <vector>
namespace xrpl::PeerFinder {
/**
* Maintains a set of IP addresses used for getting into the network.
*/
class Manager : public beast::PropertyStream::Source
{
protected:
Manager() noexcept;
public:
/**
* Destroy the object.
* Any pending source fetch operations are aborted.
* There may be some listener calls made before the
* destructor returns.
*/
~Manager() override = default;
/**
* Set the configuration for the manager.
* The new settings will be applied asynchronously.
* Thread safety:
* Can be called from any threads at any time.
*/
virtual void
setConfig(Config const& config) = 0;
/**
* Transition to the started state, synchronously.
*/
virtual void
start() = 0;
/**
* Transition to the stopped state, synchronously.
*/
virtual void
stop() = 0;
/**
* Returns the configuration for the manager.
*/
virtual Config
config() = 0;
/**
* Add a peer that should always be connected.
* This is useful for maintaining a private cluster of peers.
* The string is the name as specified in the configuration
* file, along with the set of corresponding IP addresses.
*/
virtual void
addFixedPeer(std::string_view name, std::vector<beast::IP::Endpoint> const& addresses) = 0;
/**
* Add a set of strings as fallback IP::Endpoint sources.
* @param name A label used for diagnostics.
*/
virtual void
addFallbackStrings(std::string const& name, std::vector<std::string> const& strings) = 0;
/**
* Add a URL as a fallback location to obtain IP::Endpoint sources.
* @param name A label used for diagnostics.
*/
/* VFALCO NOTE Unimplemented
virtual void addFallbackURL (std::string const& name,
std::string const& url) = 0;
*/
//--------------------------------------------------------------------------
/**
* Create a new inbound slot with the specified remote endpoint.
* If nullptr is returned, then the slot could not be assigned.
* Usually this is because of a detected self-connection.
*/
virtual std::pair<std::shared_ptr<Slot>, Result>
newInboundSlot(
beast::IP::Endpoint const& localEndpoint,
beast::IP::Endpoint const& remoteEndpoint) = 0;
/**
* Create a new outbound slot with the specified remote endpoint.
* If nullptr is returned, then the slot could not be assigned.
* Usually this is because of a duplicate connection.
*/
virtual std::pair<std::shared_ptr<Slot>, Result>
newOutboundSlot(beast::IP::Endpoint const& remoteEndpoint) = 0;
/**
* Called when mtENDPOINTS is received.
*/
virtual void
onEndpoints(std::shared_ptr<Slot> const& slot, Endpoints const& endpoints) = 0;
/**
* Called when the slot is closed.
* This always happens when the socket is closed, unless the socket
* was canceled.
*/
virtual void
onClosed(std::shared_ptr<Slot> const& slot) = 0;
/**
* Called when an outbound connection is deemed to have failed
*/
virtual void
onFailure(std::shared_ptr<Slot> const& slot) = 0;
/**
* Called when we received redirect IPs from a busy peer.
*/
virtual void
onRedirects(
boost::asio::ip::tcp::endpoint const& remoteAddress,
std::vector<boost::asio::ip::tcp::endpoint> const& eps) = 0;
//--------------------------------------------------------------------------
/**
* Called when an outbound connection attempt succeeds.
* The local endpoint must be valid. If the caller receives an error
* when retrieving the local endpoint from the socket, it should
* proceed as if the connection attempt failed by calling on_closed
* instead of on_connected.
* @return `true` if the connection should be kept
*/
virtual bool
onConnected(std::shared_ptr<Slot> const& slot, beast::IP::Endpoint const& localEndpoint) = 0;
/**
* Request an active slot type.
*/
virtual Result
activate(std::shared_ptr<Slot> const& slot, PublicKey const& key, bool reserved) = 0;
/**
* Returns a set of endpoints suitable for redirection.
*/
virtual std::vector<Endpoint>
redirect(std::shared_ptr<Slot> const& slot) = 0;
/**
* Return a set of addresses we should connect to.
*/
virtual std::vector<beast::IP::Endpoint>
autoconnect() = 0;
virtual std::vector<std::pair<std::shared_ptr<Slot>, std::vector<Endpoint>>>
buildEndpointsForPeers() = 0;
/**
* Perform periodic activity.
* This should be called once per second.
*/
virtual void
oncePerSecond() = 0;
};
} // namespace xrpl::PeerFinder

View File

@@ -0,0 +1,75 @@
#pragma once
#include <xrpl/beast/net/IPEndpoint.h>
#include <xrpl/protocol/PublicKey.h>
#include <cstdint>
#include <memory>
#include <optional>
namespace xrpl::PeerFinder {
/**
* Properties and state associated with a peer to peer overlay connection.
*/
class Slot
{
public:
using ptr = std::shared_ptr<Slot>;
enum class State { Accept, Connect, Connected, Active, Closing };
virtual ~Slot() = 0;
/**
* Returns `true` if this is an inbound connection.
*/
[[nodiscard]] virtual bool
inbound() const = 0;
/**
* Returns `true` if this is a fixed connection.
* A connection is fixed if its remote endpoint is in the list of
* remote endpoints for fixed connections.
*/
[[nodiscard]] virtual bool
fixed() const = 0;
/**
* Returns `true` if this is a reserved connection.
* It might be a cluster peer, or a peer with a reservation.
* This is only known after then handshake completes.
*/
[[nodiscard]] virtual bool
reserved() const = 0;
/**
* Returns the state of the connection.
*/
[[nodiscard]] virtual State
state() const = 0;
/**
* The remote endpoint of socket.
*/
[[nodiscard]] virtual beast::IP::Endpoint const&
remoteEndpoint() const = 0;
/**
* The local endpoint of the socket, when known.
*/
[[nodiscard]] virtual std::optional<beast::IP::Endpoint> const&
localEndpoint() const = 0;
[[nodiscard]] virtual std::optional<std::uint16_t>
listeningPort() const = 0;
/**
* The peer's public key, when known.
* The public key is established when the handshake is complete.
*/
[[nodiscard]] virtual std::optional<PublicKey> const&
publicKey() const = 0;
};
} // namespace xrpl::PeerFinder

View File

@@ -0,0 +1,46 @@
#pragma once
#include <xrpl/beast/clock/abstract_clock.h>
#include <xrpl/beast/net/IPEndpoint.h>
#include <xrpl/peerfinder/detail/Tuning.h>
#include <chrono>
#include <cstdint>
#include <vector>
namespace xrpl::PeerFinder {
using clock_type = beast::AbstractClock<std::chrono::steady_clock>;
/**
* Represents a set of addresses.
*/
using IPAddresses = std::vector<beast::IP::Endpoint>;
//------------------------------------------------------------------------------
/**
* Describes a connectable peer address along with some metadata.
*/
struct Endpoint
{
Endpoint() = default;
Endpoint(beast::IP::Endpoint ep, std::uint32_t hops);
std::uint32_t hops = 0;
beast::IP::Endpoint address;
};
inline bool
operator<(Endpoint const& lhs, Endpoint const& rhs)
{
return lhs.address < rhs.address;
}
/**
* A set of Endpoint used for connecting.
*/
using Endpoints = std::vector<Endpoint>;
} // namespace xrpl::PeerFinder

View File

@@ -0,0 +1,192 @@
#pragma once
#include <xrpl/beast/net/IPEndpoint.h>
#include <xrpl/beast/utility/Journal.h>
#include <xrpl/beast/utility/PropertyStream.h>
#include <xrpl/peerfinder/Types.h>
#include <xrpl/peerfinder/detail/Store.h>
#include <xrpl/peerfinder/detail/Tuning.h>
#include <boost/bimap.hpp>
#include <boost/bimap/multiset_of.hpp>
#include <boost/bimap/unordered_set_of.hpp>
#include <boost/iterator/transform_iterator.hpp>
#include <functional>
namespace xrpl::PeerFinder {
/**
* Stores IP addresses useful for gaining initial connections.
*
* This is one of the caches that is consulted when additional outgoing
* connections are needed. Along with the address, each entry has this
* additional metadata:
*
* Valence
* A signed integer which represents the number of successful
* consecutive connection attempts when positive, and the number of
* failed consecutive connection attempts when negative.
*
* When choosing addresses from the boot cache for the purpose of
* establishing outgoing connections, addresses are ranked in decreasing
* order of high uptime, with valence as the tie breaker.
*/
class Bootcache
{
private:
class Entry
{
public:
Entry(int valence) : valence_(valence)
{
}
int&
valence()
{
return valence_;
}
[[nodiscard]] int
valence() const
{
return valence_;
}
friend bool
operator<(Entry const& lhs, Entry const& rhs)
{
return lhs.valence() > rhs.valence();
}
private:
int valence_;
};
using left_t = boost::bimaps::
unordered_set_of<beast::IP::Endpoint, boost::hash<beast::IP::Endpoint>, std::equal_to<>>;
using right_t = boost::bimaps::multiset_of<Entry, std::less<>>;
using map_type = boost::bimap<left_t, right_t>;
using value_type = map_type::value_type;
struct Transform
{
using first_argument_type = map_type::right_map::const_iterator::value_type const&;
using result_type = beast::IP::Endpoint const&;
explicit Transform() = default;
beast::IP::Endpoint const&
operator()(map_type::right_map::const_iterator::value_type const& v) const
{
return v.get_left();
}
};
private:
map_type map_;
Store& store_;
clock_type& clock_;
beast::Journal journal_;
// Time after which we can update the database again
clock_type::time_point whenUpdate_;
// Set to true when a database update is needed
bool needsUpdate_{false};
public:
static constexpr int kStaticValence = 32;
using iterator = boost::transform_iterator<Transform, map_type::right_map::const_iterator>;
using const_iterator = iterator;
Bootcache(Store& store, clock_type& clock, beast::Journal journal);
~Bootcache();
/**
* Returns `true` if the cache is empty.
*/
[[nodiscard]] bool
empty() const;
/**
* Returns the number of entries in the cache.
*/
[[nodiscard]] map_type::size_type
size() const;
/**
* IP::Endpoint iterators that traverse in decreasing valence.
*/
/** @{ */
[[nodiscard]] const_iterator
begin() const;
[[nodiscard]] const_iterator
cbegin() const;
[[nodiscard]] const_iterator
end() const;
[[nodiscard]] const_iterator
cend() const;
void
clear();
/** @} */
/**
* Load the persisted data from the Store into the container.
*/
void
load();
/**
* Add a newly-learned address to the cache.
*/
bool
insert(beast::IP::Endpoint const& endpoint);
/**
* Add a staticallyconfigured address to the cache.
*/
bool
insertStatic(beast::IP::Endpoint const& endpoint);
/**
* Called when an outbound connection handshake completes.
*/
void
onSuccess(beast::IP::Endpoint const& endpoint);
/**
* Called when an outbound connection attempt fails to handshake.
*/
void
onFailure(beast::IP::Endpoint const& endpoint);
/**
* Stores the cache in the persistent database on a timer.
*/
void
periodicActivity();
/**
* Write the cache state to the property stream.
*/
void
onWrite(beast::PropertyStream::Map& map);
private:
void
prune();
void
update();
void
checkUpdate();
void
flagForUpdate();
};
} // namespace xrpl::PeerFinder

View File

@@ -0,0 +1,205 @@
#pragma once
#include <xrpl/beast/net/IPAddressConversion.h>
#include <xrpl/beast/net/IPEndpoint.h>
#include <boost/asio/io_context.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/intrusive/list.hpp>
#include <condition_variable>
#include <memory>
#include <mutex>
namespace xrpl::PeerFinder {
/**
* Tests remote listening sockets to make sure they are connectable.
*/
template <class Protocol = boost::asio::ip::tcp>
class Checker
{
private:
using error_code = boost::system::error_code;
struct BasicAsyncOp : boost::intrusive::list_base_hook<
boost::intrusive::link_mode<boost::intrusive::normal_link>>
{
virtual ~BasicAsyncOp() = default;
virtual void
stop() = 0;
virtual void
operator()(error_code const& ec) = 0;
};
template <class Handler>
struct AsyncOp : BasicAsyncOp
{
using socket_type = Protocol::socket;
using endpoint_type = Protocol::endpoint;
Checker& checker;
socket_type socket;
Handler handler;
AsyncOp(Checker& owner, boost::asio::io_context& ioContext, Handler&& handler);
~AsyncOp() override
{
checker.remove(*this);
}
void
stop() override;
void
operator()(error_code const& ec) override; // NOLINT(readability-identifier-naming)
};
//--------------------------------------------------------------------------
using list_type =
boost::intrusive::make_list<BasicAsyncOp, boost::intrusive::constant_time_size<true>>::type;
std::mutex mutex_;
std::condition_variable cond_;
boost::asio::io_context& ioContext_;
list_type list_;
bool stop_ = false;
public:
explicit Checker(boost::asio::io_context& ioContext);
/**
* Destroy the service.
* Any pending I/O operations will be canceled. This call blocks until
* all pending operations complete (either with success or with
* operation_aborted) and the associated thread and io_context have
* no more work remaining.
*/
~Checker();
/**
* Stop the service.
* Pending I/O operations will be canceled.
* This issues cancel orders for all pending I/O operations and then
* returns immediately. Handlers will receive operation_aborted errors,
* or if they were already queued they will complete normally.
*/
void
stop();
/**
* Block until all pending I/O completes.
*/
void
wait();
/**
* Performs an async connection test on the specified endpoint.
* The port must be non-zero. Note that the execution guarantees
* offered by asio handlers are NOT enforced.
*/
template <class Handler>
void
asyncConnect(beast::IP::Endpoint const& endpoint, Handler&& handler);
private:
void
remove(BasicAsyncOp& op);
};
//------------------------------------------------------------------------------
template <class Protocol>
template <class Handler>
Checker<Protocol>::AsyncOp<Handler>::AsyncOp(
Checker& owner,
boost::asio::io_context& ioContext,
Handler&&
handler) // NOLINT(cppcoreguidelines-rvalue-reference-param-not-moved) -- forwarded in init
: checker(owner), socket(ioContext), handler(std::forward<Handler>(handler))
{
}
template <class Protocol>
template <class Handler>
void
Checker<Protocol>::AsyncOp<Handler>::stop()
{
error_code ec;
socket.cancel(ec);
}
template <class Protocol>
template <class Handler>
void
Checker<Protocol>::AsyncOp<Handler>::operator()(error_code const& ec)
{
handler(ec);
}
//------------------------------------------------------------------------------
template <class Protocol>
Checker<Protocol>::Checker(boost::asio::io_context& ioContext) : ioContext_(ioContext)
{
}
template <class Protocol>
Checker<Protocol>::~Checker()
{
wait();
}
template <class Protocol>
void
Checker<Protocol>::stop()
{
std::scoped_lock const lock(mutex_);
if (!stop_)
{
stop_ = true;
for (auto& c : list_)
c.stop();
}
}
template <class Protocol>
void
Checker<Protocol>::wait()
{
std::unique_lock<std::mutex> lock(mutex_);
while (!list_.empty())
cond_.wait(lock);
}
template <class Protocol>
template <class Handler>
void
Checker<Protocol>::asyncConnect(beast::IP::Endpoint const& endpoint, Handler&& handler)
{
auto const op =
std::make_shared<AsyncOp<Handler>>(*this, ioContext_, std::forward<Handler>(handler));
{
std::scoped_lock const lock(mutex_);
list_.push_back(*op);
}
op->socket.async_connect(
beast::IPAddressConversion::toAsioEndpoint(endpoint),
[op](error_code const& ec) { (*op)(ec); });
}
template <class Protocol>
void
Checker<Protocol>::remove(BasicAsyncOp& op)
{
std::scoped_lock const lock(mutex_);
list_.erase(list_.iterator_to(op));
if (list_.size() == 0)
cond_.notify_all();
}
} // namespace xrpl::PeerFinder

View File

@@ -0,0 +1,394 @@
#pragma once
#include <xrpl/beast/utility/PropertyStream.h>
#include <xrpl/beast/utility/instrumentation.h>
#include <xrpl/peerfinder/Config.h>
#include <xrpl/peerfinder/Slot.h>
#include <xrpl/peerfinder/detail/Tuning.h>
#include <cstddef>
#include <sstream>
#include <string>
namespace xrpl::PeerFinder {
/**
* Direction of a slot count adjustment.
*/
enum class CountAdjustment : int { Decrement = -1, Increment = 1 };
/**
* Manages the count of available connections for the various slots.
*/
class Counts
{
public:
/**
* Adds the slot state and properties to the slot counts.
*/
void
add(Slot const& s)
{
adjust(s, CountAdjustment::Increment);
}
/**
* Removes the slot state and properties from the slot counts.
*/
void
remove(Slot const& s)
{
adjust(s, CountAdjustment::Decrement);
}
/**
* Returns `true` if the slot can become active.
*/
[[nodiscard]] bool
canActivate(Slot const& s) const
{
// Must be handshaked and in the right state
XRPL_ASSERT(
s.state() == Slot::State::Connected || s.state() == Slot::State::Accept,
"xrpl::PeerFinder::Counts::can_activate : valid input state");
if (s.fixed() || s.reserved())
return true;
if (s.inbound())
return inActive_ < inMax_;
return outActive_ < outMax_;
}
/**
* Returns the number of attempts needed to bring us to the max.
*/
[[nodiscard]] std::size_t
attemptsNeeded() const
{
if (attempts_ >= Tuning::kMaxConnectAttempts)
return 0;
return Tuning::kMaxConnectAttempts - attempts_;
}
/**
* Returns the number of outbound connection attempts.
*/
[[nodiscard]] std::size_t
attempts() const
{
return attempts_;
}
/**
* Returns the total number of outbound slots.
*/
[[nodiscard]] int
outMax() const
{
return outMax_;
}
/**
* Returns the number of outbound peers assigned an open slot.
* Fixed peers do not count towards outbound slots used.
*/
[[nodiscard]] int
outActive() const
{
return outActive_;
}
/**
* Returns the number of fixed connections.
*/
[[nodiscard]] std::size_t
fixed() const
{
return fixed_;
}
/**
* Returns the number of active fixed connections.
*/
[[nodiscard]] std::size_t
fixedActive() const
{
return fixedActive_;
}
//--------------------------------------------------------------------------
/**
* Called when the config is set or changed.
*/
void
onConfig(Config const& config)
{
outMax_ = config.outPeers;
if (config.wantIncoming)
inMax_ = config.inPeers;
}
/**
* Returns the number of accepted connections that haven't handshaked.
*/
[[nodiscard]] int
acceptCount() const
{
return acceptCount_;
}
/**
* Returns the number of connection attempts currently active.
*/
[[nodiscard]] int
connectCount() const
{
return attempts_;
}
/**
* Returns the number of connections that are gracefully closing.
*/
[[nodiscard]] int
closingCount() const
{
return closingCount_;
}
/**
* Returns the total number of inbound slots.
*/
[[nodiscard]] int
inMax() const
{
return inMax_;
}
/**
* Returns the number of inbound peers assigned an open slot.
*/
[[nodiscard]] int
inboundActive() const
{
return inActive_;
}
/**
* Returns the total number of active peers excluding fixed peers.
*/
[[nodiscard]] int
totalActive() const
{
return inActive_ + outActive_;
}
/**
* Returns the number of unused inbound slots.
* Fixed peers do not deduct from inbound slots or count towards totals.
*/
[[nodiscard]] int
inboundSlotsFree() const
{
if (inActive_ < inMax_)
return inMax_ - inActive_;
return 0;
}
/**
* Returns the number of unused outbound slots.
* Fixed peers do not deduct from outbound slots or count towards totals.
*/
[[nodiscard]] int
outboundSlotsFree() const
{
if (outActive_ < outMax_)
return outMax_ - outActive_;
return 0;
}
//--------------------------------------------------------------------------
/**
* Returns true if the slot logic considers us "connected" to the network.
*/
[[nodiscard]] bool
isConnectedToNetwork() const
{
// We will consider ourselves connected if we have reached
// the number of outgoing connections desired, or if connect
// automatically is false.
//
// Fixed peers do not count towards the active outgoing total.
return outMax_ <= 0;
}
/**
* Output statistics.
*/
void
onWrite(beast::PropertyStream::Map& map) const
{
map["accept"] = acceptCount();
map["connect"] = connectCount();
map["close"] = closingCount();
map["in"] << inActive_ << "/" << inMax_;
map["out"] << outActive_ << "/" << outMax_;
map["fixed"] = fixedActive_;
map["reserved"] = reserved_;
map["total"] = active_;
}
/**
* Records the state for diagnostics.
*/
[[nodiscard]] std::string
stateString() const
{
std::stringstream ss;
ss << outActive_ << "/" << outMax_ << " out, " << inActive_ << "/" << inMax_ << " in, "
<< connectCount() << " connecting, " << closingCount() << " closing";
return ss.str();
}
//--------------------------------------------------------------------------
private:
/**
* Increments or decrements a counter based on the adjustment direction.
*/
template <typename T>
static void
adjustCounter(T& counter, CountAdjustment dir)
{
switch (dir)
{
case CountAdjustment::Increment:
++counter;
break;
case CountAdjustment::Decrement:
--counter;
break;
}
}
// Adjusts counts based on the specified slot, in the direction indicated.
//
// IMPORTANT: All std::size_t counters MUST be adjusted via adjustCounter()
// and NEVER via `+= n` where n = static_cast<int>(dir). When dir is
// Decrement, n == -1; adding -1 to a std::size_t implicitly converts -1 to
// SIZE_MAX, which UBSan flags as unsigned-integer-overflow and masks real
// underflow bugs (decrementing a counter already at zero). Plain int
// counters (acceptCount_, attempts_, closingCount_) are safe with += n.
void
adjust(Slot const& s, CountAdjustment const dir)
{
int const n = static_cast<int>(dir);
if (s.fixed())
adjustCounter(fixed_, dir);
if (s.reserved())
adjustCounter(reserved_, dir);
switch (s.state())
{
case Slot::State::Accept:
XRPL_ASSERT(s.inbound(), "xrpl::PeerFinder::Counts::adjust : input is inbound");
acceptCount_ += n;
break;
case Slot::State::Connect:
case Slot::State::Connected:
XRPL_ASSERT(
!s.inbound(),
"xrpl::PeerFinder::Counts::adjust : input is not "
"inbound");
attempts_ += n;
break;
case Slot::State::Active:
if (s.fixed())
adjustCounter(fixedActive_, dir);
if (!s.fixed() && !s.reserved())
{
if (s.inbound())
{
adjustCounter(inActive_, dir);
}
else
{
adjustCounter(outActive_, dir);
}
}
adjustCounter(active_, dir);
break;
case Slot::State::Closing:
closingCount_ += n;
break;
// LCOV_EXCL_START
default:
UNREACHABLE("xrpl::PeerFinder::Counts::adjust : invalid input state");
break;
// LCOV_EXCL_STOP
};
}
private:
/**
* Outbound connection attempts.
*/
int attempts_{0};
/**
* Active connections, including fixed and reserved.
*/
std::size_t active_{0};
/**
* Total number of inbound slots.
*/
std::size_t inMax_{0};
/**
* Number of inbound slots assigned to active peers.
*/
std::size_t inActive_{0};
/**
* Maximum desired outbound slots.
*/
std::size_t outMax_{0};
/**
* Active outbound slots.
*/
std::size_t outActive_{0};
/**
* Fixed connections.
*/
std::size_t fixed_{0};
/**
* Active fixed connections.
*/
std::size_t fixedActive_{0};
/**
* Reserved connections.
*/
std::size_t reserved_{0};
// Number of inbound connections that are
// not active or gracefully closing.
int acceptCount_{0};
// Number of connections that are gracefully closing.
int closingCount_{0};
};
} // namespace xrpl::PeerFinder

View File

@@ -0,0 +1,58 @@
#pragma once
#include <xrpl/peerfinder/Types.h>
#include <xrpl/peerfinder/detail/Tuning.h>
#include <algorithm>
#include <chrono>
#include <cstddef>
namespace xrpl::PeerFinder {
/**
* Metadata for a Fixed slot.
*/
class Fixed
{
public:
explicit Fixed(clock_type& clock) : when_(clock.now())
{
}
Fixed(Fixed const&) = default;
/**
* Returns the time after which we should allow a connection attempt.
*/
[[nodiscard]] clock_type::time_point const&
when() const
{
return when_;
}
/**
* Updates metadata to reflect a failed connection.
*/
void
failure(clock_type::time_point const& now)
{
failures_ = std::min(failures_ + 1, Tuning::kConnectionBackoff.size() - 1);
when_ = now + std::chrono::minutes(Tuning::kConnectionBackoff[failures_]);
}
/**
* Updates metadata to reflect a successful connection.
*/
void
success(clock_type::time_point const& now)
{
failures_ = 0;
when_ = now;
}
private:
clock_type::time_point when_;
std::size_t failures_{0};
};
} // namespace xrpl::PeerFinder

View File

@@ -0,0 +1,344 @@
#pragma once
#include <xrpl/beast/container/aged_set.h>
#include <xrpl/beast/net/IPAddress.h>
#include <xrpl/beast/utility/instrumentation.h>
#include <xrpl/peerfinder/Types.h>
#include <xrpl/peerfinder/detail/SlotImp.h>
#include <xrpl/peerfinder/detail/Tuning.h>
#include <algorithm>
#include <cstddef>
#include <utility>
#include <vector>
namespace xrpl::PeerFinder {
namespace detail {
/**
* Try to insert one object in the target.
* When an item is handed out it is moved to the end of the container.
* @return The number of objects inserted
*/
// VFALCO TODO specialization that handles std::list for SequenceContainer
// using splice for optimization over erase/push_back
//
template <class Target, class HopContainer>
std::size_t
handoutOne(Target& t, HopContainer& h)
{
XRPL_ASSERT(!t.full(), "xrpl::PeerFinder::detail::handoutOne : target is not full");
for (auto it = h.begin(); it != h.end(); ++it)
{
auto const& e = *it;
if (t.tryInsert(e))
{
h.moveBack(it);
return 1;
}
}
return 0;
}
} // namespace detail
/**
* Distributes objects to targets according to business rules.
* A best effort is made to evenly distribute items in the sequence
* container list into the target sequence list.
*/
template <class TargetFwdIter, class SeqFwdIter>
void
handout(TargetFwdIter first, TargetFwdIter last, SeqFwdIter seqFirst, SeqFwdIter seqLast)
{
for (;;)
{
std::size_t n(0);
for (auto si = seqFirst; si != seqLast; ++si)
{
auto c = *si;
bool allFull(true);
for (auto ti = first; ti != last; ++ti)
{
auto& t = *ti;
if (!t.full())
{
n += detail::handoutOne(t, c);
allFull = false;
}
}
if (allFull)
return;
}
if (!n)
break;
}
}
//------------------------------------------------------------------------------
/**
* Receives handouts for redirecting a connection.
* An incoming connection request is redirected when we are full on slots.
*/
class RedirectHandouts
{
public:
template <class = void>
explicit RedirectHandouts(SlotImp::ptr slot);
template <class = void>
bool
tryInsert(Endpoint const& ep);
[[nodiscard]] bool
full() const
{
return list_.size() >= Tuning::kRedirectEndpointCount;
}
[[nodiscard]] SlotImp::ptr const&
slot() const
{
return slot_;
}
std::vector<Endpoint>&
list()
{
return list_;
}
[[nodiscard]] std::vector<Endpoint> const&
list() const
{
return list_;
}
private:
SlotImp::ptr slot_;
std::vector<Endpoint> list_;
};
template <class>
RedirectHandouts::RedirectHandouts(SlotImp::ptr slot) : slot_(std::move(slot))
{
list_.reserve(Tuning::kRedirectEndpointCount);
}
template <class>
bool
RedirectHandouts::tryInsert(Endpoint const& ep)
{
if (full())
return false;
// VFALCO NOTE This check can be removed when we provide the
// addresses in a peer HTTP handshake instead of
// the tmENDPOINTS message.
//
if (ep.hops > Tuning::kMaxHops)
return false;
// Don't send them our address
if (ep.hops == 0)
return false;
// Don't send them their own address
if (slot_->remoteEndpoint().address() == ep.address.address())
return false;
// Make sure the address isn't already in our list
if (std::ranges::any_of(list_, [&ep](Endpoint const& other) {
// Ignore port for security reasons
return other.address.address() == ep.address.address();
}))
{
return false;
}
list_.emplace_back(ep.address, ep.hops);
return true;
}
//------------------------------------------------------------------------------
/**
* Receives endpoints for a slot during periodic handouts.
*/
class SlotHandouts
{
public:
template <class = void>
explicit SlotHandouts(SlotImp::ptr slot);
template <class = void>
bool
tryInsert(Endpoint const& ep);
[[nodiscard]] bool
full() const
{
return list_.size() >= Tuning::kNumberOfEndpoints;
}
void
insert(Endpoint const& ep)
{
list_.push_back(ep);
}
[[nodiscard]] SlotImp::ptr const&
slot() const
{
return slot_;
}
[[nodiscard]] std::vector<Endpoint> const&
list() const
{
return list_;
}
private:
SlotImp::ptr slot_;
std::vector<Endpoint> list_;
};
template <class>
SlotHandouts::SlotHandouts(SlotImp::ptr slot) : slot_(std::move(slot))
{
list_.reserve(Tuning::kNumberOfEndpoints);
}
template <class>
bool
SlotHandouts::tryInsert(Endpoint const& ep)
{
if (full())
return false;
if (ep.hops > Tuning::kMaxHops)
return false;
if (slot_->recent.filter(ep.address, ep.hops))
return false;
// Don't send them their own address
if (slot_->remoteEndpoint().address() == ep.address.address())
return false;
// Make sure the address isn't already in our list
if (std::ranges::any_of(list_, [&ep](Endpoint const& other) {
// Ignore port for security reasons
return other.address.address() == ep.address.address();
}))
return false;
list_.emplace_back(ep.address, ep.hops);
// Insert into this slot's recent table. Although the endpoint
// didn't come from the slot, adding it to the slot's table
// prevents us from sending it again until it has expired from
// the other end's cache.
//
slot_->recent.insert(ep.address, ep.hops);
return true;
}
//------------------------------------------------------------------------------
/**
* Receives handouts for making automatic connections.
*/
class ConnectHandouts
{
public:
// Keeps track of addresses we have made outgoing connections
// to, for the purposes of not connecting to them too frequently.
using Squelches = beast::aged_set<beast::IP::Address>;
using list_type = std::vector<beast::IP::Endpoint>;
private:
std::size_t needed_;
Squelches& squelches_;
list_type list_;
public:
template <class = void>
ConnectHandouts(std::size_t needed, Squelches& squelches);
template <class = void>
bool
tryInsert(beast::IP::Endpoint const& endpoint);
[[nodiscard]] bool
empty() const
{
return list_.empty();
}
[[nodiscard]] bool
full() const
{
return list_.size() >= needed_;
}
bool
tryInsert(Endpoint const& endpoint)
{
return tryInsert(endpoint.address);
}
list_type&
list()
{
return list_;
}
[[nodiscard]] list_type const&
list() const
{
return list_;
}
};
template <class>
ConnectHandouts::ConnectHandouts(std::size_t needed, Squelches& squelches)
: needed_(needed), squelches_(squelches)
{
list_.reserve(needed);
}
template <class>
bool
ConnectHandouts::tryInsert(beast::IP::Endpoint const& endpoint)
{
if (full())
return false;
// Make sure the address isn't already in our list
if (std::ranges::any_of(list_, [&endpoint](beast::IP::Endpoint const& other) {
// Ignore port for security reasons
return other.address() == endpoint.address();
}))
{
return false;
}
// Add to squelch list so we don't try it too often.
// If its already there, then make try_insert fail.
auto const result(squelches_.insert(endpoint.address()));
if (!result.second)
return false;
list_.push_back(endpoint);
return true;
}
} // namespace xrpl::PeerFinder

View File

@@ -0,0 +1,564 @@
#pragma once
#include <xrpl/basics/Log.h>
#include <xrpl/basics/random.h>
#include <xrpl/beast/container/aged_map.h>
#include <xrpl/beast/net/IPEndpoint.h>
#include <xrpl/beast/utility/Journal.h>
#include <xrpl/beast/utility/PropertyStream.h>
#include <xrpl/beast/utility/instrumentation.h>
#include <xrpl/beast/utility/maybe_const.h>
#include <xrpl/peerfinder/Types.h>
#include <xrpl/peerfinder/detail/Tuning.h>
#include <boost/intrusive/list.hpp>
#include <boost/iterator/transform_iterator.hpp>
#include <algorithm>
#include <array>
#include <chrono>
#include <cstddef>
#include <cstdint>
#include <functional>
#include <iomanip>
#include <ios>
#include <iterator>
#include <memory>
#include <sstream>
#include <string>
#include <utility>
#include <vector>
namespace xrpl::PeerFinder {
template <class>
class Livecache;
namespace detail {
class LivecacheBase
{
public:
explicit LivecacheBase() = default;
protected:
struct Element : boost::intrusive::list_base_hook<>
{
Element(Endpoint endpoint) : endpoint(std::move(endpoint))
{
}
Endpoint endpoint;
};
using list_type =
boost::intrusive::make_list<Element, boost::intrusive::constant_time_size<false>>::type;
public:
/**
* A list of Endpoint at the same hops
* This is a lightweight wrapper around a reference to the underlying
* container.
*/
template <bool IsConst>
class Hop
{
public:
// Iterator transformation to extract the endpoint from Element
struct Transform
{
using first_argument = Element;
using result_type = Endpoint;
explicit Transform() = default;
Endpoint const&
operator()(Element const& e) const
{
return e.endpoint;
}
};
public:
using iterator = boost::transform_iterator<Transform, list_type::const_iterator>;
using const_iterator = iterator;
using reverse_iterator =
boost::transform_iterator<Transform, list_type::const_reverse_iterator>;
using const_reverse_iterator = reverse_iterator;
[[nodiscard]] iterator
begin() const
{
return iterator(list_.get().cbegin(), Transform());
}
[[nodiscard]] iterator
cbegin() const
{
return iterator(list_.get().cbegin(), Transform());
}
[[nodiscard]] iterator
end() const
{
return iterator(list_.get().cend(), Transform());
}
[[nodiscard]] iterator
cend() const
{
return iterator(list_.get().cend(), Transform());
}
[[nodiscard]] reverse_iterator
rbegin() const
{
return reverse_iterator(list_.get().crbegin(), Transform());
}
[[nodiscard]] reverse_iterator
crbegin() const
{
return reverse_iterator(list_.get().crbegin(), Transform());
}
[[nodiscard]] reverse_iterator
rend() const
{
return reverse_iterator(list_.get().crend(), Transform());
}
[[nodiscard]] reverse_iterator
crend() const
{
return reverse_iterator(list_.get().crend(), Transform());
}
// move the element to the end of the container
void
moveBack(const_iterator pos)
{
auto& e(const_cast<Element&>(*pos.base()));
list_.get().erase(list_.get().iterator_to(e));
list_.get().push_back(e);
}
private:
explicit Hop(beast::MaybeConst<IsConst, list_type>::type& list) : list_(list)
{
}
friend class LivecacheBase;
std::reference_wrapper<typename beast::MaybeConst<IsConst, list_type>::type> list_;
};
protected:
// Work-around to call Hop's private constructor from Livecache
template <bool IsConst>
static Hop<IsConst>
makeHop(beast::MaybeConst<IsConst, list_type>::type& list)
{
return Hop<IsConst>(list);
}
};
} // namespace detail
//------------------------------------------------------------------------------
/**
* The Livecache holds the short-lived relayed Endpoint messages.
*
* Since peers only advertise themselves when they have open slots,
* we want these messages to expire rather quickly after the peer becomes
* full.
*
* Addresses added to the cache are not connection-tested to see if
* they are connectable (with one small exception regarding neighbors).
* Therefore, these addresses are not suitable for persisting across
* launches or for bootstrapping, because they do not have verifiable
* and locally observed uptime and connectability information.
*/
template <class Allocator = std::allocator<char>>
class Livecache : protected detail::LivecacheBase
{
private:
using cache_type = beast::aged_map<
beast::IP::Endpoint,
Element,
std::chrono::steady_clock,
std::less<beast::IP::Endpoint>,
Allocator>;
beast::Journal journal_;
cache_type cache_;
public:
using allocator_type = Allocator;
/**
* Create the cache.
*/
Livecache(clock_type& clock, beast::Journal journal, Allocator alloc = Allocator());
//
// Iteration by hops
//
// The range [begin, end) provides a sequence of list_type
// where each list contains endpoints at a given hops.
//
class HopsT
{
private:
// An endpoint at hops=0 represents the local node.
// Endpoints coming in at maxHops are stored at maxHops +1,
// but not given out (since they would exceed maxHops). They
// are used for automatic connection attempts.
//
using Histogram = std::array<int, 1 + Tuning::kMaxHops + 1>;
using lists_type = std::array<list_type, 1 + Tuning::kMaxHops + 1>;
template <bool IsConst>
struct Transform
{
using first_argument = lists_type::value_type;
using result_type = Hop<IsConst>;
explicit Transform() = default;
Hop<IsConst>
operator()(beast::MaybeConst<IsConst, lists_type::value_type>::type& list) const
{
return makeHop<IsConst>(list);
}
};
public:
using iterator = boost::transform_iterator<Transform<false>, lists_type::iterator>;
using const_iterator =
boost::transform_iterator<Transform<true>, lists_type::const_iterator>;
using reverse_iterator =
boost::transform_iterator<Transform<false>, lists_type::reverse_iterator>;
using const_reverse_iterator =
boost::transform_iterator<Transform<true>, lists_type::const_reverse_iterator>;
iterator
begin()
{
return iterator(lists_.begin(), Transform<false>());
}
[[nodiscard]] const_iterator
begin() const
{
return const_iterator(lists_.cbegin(), Transform<true>());
}
[[nodiscard]] const_iterator
cbegin() const
{
return const_iterator(lists_.cbegin(), Transform<true>());
}
iterator
end()
{
return iterator(lists_.end(), Transform<false>());
}
[[nodiscard]] const_iterator
end() const
{
return const_iterator(lists_.cend(), Transform<true>());
}
[[nodiscard]] const_iterator
cend() const
{
return const_iterator(lists_.cend(), Transform<true>());
}
reverse_iterator
rbegin()
{
return reverse_iterator(lists_.rbegin(), Transform<false>());
}
[[nodiscard]] const_reverse_iterator
rbegin() const
{
return const_reverse_iterator(lists_.crbegin(), Transform<true>());
}
[[nodiscard]] const_reverse_iterator
crbegin() const
{
return const_reverse_iterator(lists_.crbegin(), Transform<true>());
}
reverse_iterator
rend()
{
return reverse_iterator(lists_.rend(), Transform<false>());
}
[[nodiscard]] const_reverse_iterator
rend() const
{
return const_reverse_iterator(lists_.crend(), Transform<true>());
}
[[nodiscard]] const_reverse_iterator
crend() const
{
return const_reverse_iterator(lists_.crend(), Transform<true>());
}
/**
* Shuffle each hop list.
*/
void
shuffle();
[[nodiscard]] std::string
histogram() const;
private:
explicit HopsT(Allocator const& alloc);
void
insert(Element& e);
// Reinsert e at a new hops
void
reinsert(Element& e, std::uint32_t hops);
void
remove(Element& e);
friend class Livecache;
lists_type lists_;
Histogram hist_{};
} hops;
/**
* Returns `true` if the cache is empty.
*/
[[nodiscard]] bool
empty() const
{
return cache_.empty();
}
/**
* Returns the number of entries in the cache.
*/
cache_type::size_type
size() const
{
return cache_.size();
}
/**
* Erase entries whose time has expired.
*/
void
expire();
/**
* Creates or updates an existing Element based on a new message.
*/
void
insert(Endpoint const& ep);
/**
* Output statistics.
*/
void
onWrite(beast::PropertyStream::Map& map);
};
//------------------------------------------------------------------------------
template <class Allocator>
Livecache<Allocator>::Livecache(clock_type& clock, beast::Journal journal, Allocator alloc)
: journal_(journal), cache_(clock, alloc), hops(alloc)
{
}
template <class Allocator>
void
Livecache<Allocator>::expire()
{
std::size_t n(0);
typename cache_type::time_point const expired(
cache_.clock().now() - Tuning::kLiveCacheSecondsToLive);
for (auto iter(cache_.chronological.begin());
iter != cache_.chronological.end() && iter.when() <= expired;)
{
Element& e(iter->second);
hops.remove(e);
iter = cache_.erase(iter);
++n;
}
if (n > 0)
{
JLOG(journal_.debug()) << std::left << std::setw(18) << "Livecache expired " << n
<< ((n > 1) ? " entries" : " entry");
}
}
template <class Allocator>
void
Livecache<Allocator>::insert(Endpoint const& ep)
{
// The caller already incremented hop, so if we got a
// message at maxHops we will store it at maxHops + 1.
// This means we won't give out the address to other peers
// but we will use it to make connections and hand it out
// when redirecting.
//
XRPL_ASSERT(
ep.hops <= (Tuning::kMaxHops + 1),
"xrpl::PeerFinder::Livecache::insert : maximum input hops");
auto result = cache_.emplace(ep.address, ep);
Element& e(result.first->second);
if (result.second)
{
hops.insert(e);
JLOG(journal_.debug()) << std::left << std::setw(18) << "Livecache insert " << ep.address
<< " at hops " << ep.hops;
return;
}
if (!result.second && (ep.hops > e.endpoint.hops))
{
// Drop duplicates at higher hops
std::size_t const excess(ep.hops - e.endpoint.hops);
JLOG(journal_.trace()) << std::left << std::setw(18) << "Livecache drop " << ep.address
<< " at hops +" << excess;
return;
}
cache_.touch(result.first);
// Address already in the cache so update metadata
if (ep.hops < e.endpoint.hops)
{
hops.reinsert(e, ep.hops);
JLOG(journal_.debug()) << std::left << std::setw(18) << "Livecache update " << ep.address
<< " at hops " << ep.hops;
}
else
{
JLOG(journal_.trace()) << std::left << std::setw(18) << "Livecache refresh " << ep.address
<< " at hops " << ep.hops;
}
}
template <class Allocator>
void
Livecache<Allocator>::onWrite(beast::PropertyStream::Map& map)
{
typename cache_type::time_point const expired(
cache_.clock().now() - Tuning::kLiveCacheSecondsToLive);
map["size"] = size();
map["hist"] = hops.histogram();
beast::PropertyStream::Set set("entries", map);
for (auto iter(cache_.cbegin()); iter != cache_.cend(); ++iter)
{
auto const& e(iter->second);
beast::PropertyStream::Map item(set);
item["hops"] = e.endpoint.hops;
item["address"] = e.endpoint.address.toString();
std::stringstream ss;
ss << (iter.when() - expired).count();
item["expires"] = ss.str();
}
}
//------------------------------------------------------------------------------
template <class Allocator>
void
Livecache<Allocator>::HopsT::shuffle()
{
for (auto& list : lists_)
{
std::vector<std::reference_wrapper<Element>> v;
v.reserve(list.size());
std::ranges::copy(list, std::back_inserter(v));
std::shuffle(v.begin(), v.end(), defaultPrng());
list.clear();
for (auto& e : v)
list.push_back(e);
}
}
template <class Allocator>
std::string
Livecache<Allocator>::HopsT::histogram() const
{
std::string s;
for (auto const& h : hist_)
{
if (!s.empty())
s += ", ";
s += std::to_string(h);
}
return s;
}
template <class Allocator>
Livecache<Allocator>::HopsT::HopsT(Allocator const& alloc)
{
std::ranges::fill(hist_, 0);
}
template <class Allocator>
void
Livecache<Allocator>::HopsT::insert(Element& e)
{
XRPL_ASSERT(
e.endpoint.hops <= Tuning::kMaxHops + 1,
"xrpl::PeerFinder::Livecache::HopsT::insert : maximum input hops");
// This has security implications without a shuffle
lists_[e.endpoint.hops].push_front(e);
++hist_[e.endpoint.hops];
}
template <class Allocator>
void
Livecache<Allocator>::HopsT::reinsert(Element& e, std::uint32_t numHops)
{
XRPL_ASSERT(
numHops <= Tuning::kMaxHops + 1,
"xrpl::PeerFinder::Livecache::HopsT::reinsert : maximum hops input");
auto& list = lists_[e.endpoint.hops];
list.erase(list.iterator_to(e));
--hist_[e.endpoint.hops];
e.endpoint.hops = numHops;
insert(e);
}
template <class Allocator>
void
Livecache<Allocator>::HopsT::remove(Element& e)
{
--hist_[e.endpoint.hops];
auto& list = lists_[e.endpoint.hops];
list.erase(list.iterator_to(e));
}
} // namespace xrpl::PeerFinder

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,199 @@
#pragma once
#include <xrpl/beast/container/aged_unordered_map.h>
#include <xrpl/beast/net/IPEndpoint.h>
#include <xrpl/peerfinder/Slot.h>
#include <xrpl/peerfinder/Types.h>
#include <xrpl/protocol/PublicKey.h>
#include <atomic>
#include <cstdint>
#include <memory>
#include <optional>
#include <string>
namespace xrpl::PeerFinder {
class SlotImp : public Slot
{
public:
using ptr = std::shared_ptr<SlotImp>;
// inbound
SlotImp(
beast::IP::Endpoint const& localEndpoint,
beast::IP::Endpoint remoteEndpoint,
bool fixed,
clock_type& clock);
// outbound
SlotImp(beast::IP::Endpoint remoteEndpoint, bool fixed, clock_type& clock);
bool
inbound() const override
{
return inbound_;
}
bool
fixed() const override
{
return fixed_;
}
bool
reserved() const override
{
return reserved_;
}
State
state() const override
{
return state_;
}
beast::IP::Endpoint const&
remoteEndpoint() const override
{
return remoteEndpoint_;
}
std::optional<beast::IP::Endpoint> const&
localEndpoint() const override
{
return localEndpoint_;
}
std::optional<PublicKey> const&
publicKey() const override
{
return publicKey_;
}
std::string
prefix() const
{
return "[" + getFingerprint(remoteEndpoint(), publicKey()) + "] ";
}
std::optional<std::uint16_t>
listeningPort() const override
{
std::uint32_t const value = listeningPort_;
if (value == kUnknownPort)
return std::nullopt;
return value;
}
void
setListeningPort(std::uint16_t port)
{
listeningPort_ = port;
}
void
localEndpoint(beast::IP::Endpoint const& endpoint)
{
localEndpoint_ = endpoint;
}
void
remoteEndpoint(beast::IP::Endpoint const& endpoint)
{
remoteEndpoint_ = endpoint;
}
void
publicKey(PublicKey const& key)
{
publicKey_ = key;
}
void
reserved(bool reserved)
{
reserved_ = reserved;
}
//--------------------------------------------------------------------------
void
state(State state);
void
activate(clock_type::time_point const& now);
// "Memberspace"
//
// The set of all recent addresses that we have seen from this peer.
// We try to avoid sending a peer the same addresses they gave us.
//
class RecentT
{
public:
explicit RecentT(clock_type& clock);
/**
* Called for each valid endpoint received for a slot.
* We also insert messages that we send to the slot to prevent
* sending a slot the same address too frequently.
*/
void
insert(beast::IP::Endpoint const& ep, std::uint32_t hops);
/**
* Returns `true` if we should not send endpoint to the slot.
*/
bool
filter(beast::IP::Endpoint const& ep, std::uint32_t hops);
private:
void
expire();
friend class SlotImp;
beast::aged_unordered_map<beast::IP::Endpoint, std::uint32_t> cache_;
} recent;
void
expire()
{
recent.expire();
}
private:
bool const inbound_;
bool const fixed_;
bool reserved_;
State state_;
beast::IP::Endpoint remoteEndpoint_;
std::optional<beast::IP::Endpoint> localEndpoint_;
std::optional<PublicKey> publicKey_;
static std::int32_t constexpr kUnknownPort = -1;
std::atomic<std::int32_t> listeningPort_;
public:
// DEPRECATED public data members
// Tells us if we checked the connection. Outbound connections
// are always considered checked since we successfully connected.
bool checked;
// Set to indicate if the connection can receive incoming at the
// address advertised in mtENDPOINTS. Only valid if checked is true.
bool canAccept;
// Set to indicate that a connection check for this peer is in
// progress. Valid always.
bool connectivityCheckInProgress;
// The time after which we will accept mtENDPOINTS from the peer
// This is to prevent flooding or spamming. Receipt of mtENDPOINTS
// sooner than the allotted time should impose a load charge.
//
clock_type::time_point whenAcceptEndpoints;
};
} // namespace xrpl::PeerFinder

View File

@@ -0,0 +1,49 @@
#pragma once
#include <xrpl/beast/utility/Journal.h>
#include <xrpl/peerfinder/Types.h>
#include <boost/system/error_code.hpp>
#include <string>
namespace xrpl::PeerFinder {
/**
* A static or dynamic source of peer addresses.
* These are used as fallbacks when we are bootstrapping and don't have
* a local cache, or when none of our addresses are functioning. Typically
* sources will represent things like static text in the config file, a
* separate local file with addresses, or a remote HTTPS URL that can
* be updated automatically. Another solution is to use a custom DNS server
* that hands out peer IP addresses when name lookups are performed.
*/
class Source
{
public:
/**
* The results of a fetch.
*/
struct Results
{
explicit Results() = default;
// error_code on a failure
boost::system::error_code error;
// list of fetched endpoints
IPAddresses addresses;
};
virtual ~Source() = default;
virtual std::string const&
name() = 0;
virtual void
cancel()
{
}
virtual void
fetch(Results& results, beast::Journal journal) = 0;
};
} // namespace xrpl::PeerFinder

View File

@@ -0,0 +1,25 @@
#pragma once
#include <xrpl/peerfinder/detail/Source.h>
#include <memory>
#include <string>
#include <vector>
namespace xrpl::PeerFinder {
/**
* Provides addresses from a static set of strings.
*/
class SourceStrings : public Source
{
public:
explicit SourceStrings() = default;
using Strings = std::vector<std::string>;
static std::shared_ptr<Source>
make(std::string const& name, Strings const& strings);
};
} // namespace xrpl::PeerFinder

View File

@@ -0,0 +1,36 @@
#pragma once
#include <xrpl/beast/net/IPEndpoint.h>
#include <cstddef>
#include <functional>
#include <vector>
namespace xrpl::PeerFinder {
/**
* Abstract persistence for PeerFinder data.
*/
class Store
{
public:
virtual ~Store() = default;
// load the bootstrap cache
using load_callback = std::function<void(beast::IP::Endpoint, int)>;
virtual std::size_t
load(load_callback const& cb) = 0;
// save the bootstrap cache
struct Entry
{
explicit Entry() = default;
beast::IP::Endpoint endpoint;
int valence{};
};
virtual void
save(std::vector<Entry> const& v) = 0;
};
} // namespace xrpl::PeerFinder

View File

@@ -0,0 +1,115 @@
#pragma once
#include <algorithm>
#include <array>
#include <chrono>
#include <cstdint>
/**
* Heuristically tuned constants.
*/
/** @{ */
namespace xrpl::PeerFinder::Tuning {
//---------------------------------------------------------
//
// Automatic Connection Policy
//
//---------------------------------------------------------
/**
* Time to wait between making batches of connection attempts
*/
static constexpr auto kSecondsPerConnect = 10;
/**
* Maximum number of simultaneous connection attempts.
*/
static constexpr auto kMaxConnectAttempts = 20;
/**
* The percentage of total peer slots that are outbound.
* The number of outbound peers will be the larger of the
* minOutCount and outPercent * Config::maxPeers specially
* rounded.
*/
static constexpr auto kOutPercent = 15;
/**
* A hard minimum on the number of outgoing connections.
* This is enforced outside the Logic, so that the unit test
* can use any settings it wants.
*/
static constexpr auto kMinOutCount = 10;
/**
* The default value of Config::maxPeers.
*/
static constexpr auto kDefaultMaxPeers = 21;
/**
* Max redirects we will accept from one connection.
* Redirects are limited for security purposes, to prevent
* the address caches from getting flooded.
*/
static constexpr auto kMaxRedirects = 30;
//------------------------------------------------------------------------------
//
// Fixed
//
//------------------------------------------------------------------------------
static std::array<int, 10> const kConnectionBackoff{{1, 1, 2, 3, 5, 8, 13, 21, 34, 55}};
//------------------------------------------------------------------------------
//
// Bootcache
//
//------------------------------------------------------------------------------
// Threshold of cache entries above which we trim.
static constexpr auto kBootcacheSize = 1000;
// The percentage of addresses we prune when we trim the cache.
static constexpr auto kBootcachePrunePercent = 10;
// The cool down wait between database updates
// Ideally this should be larger than the time it takes a full
// peer to send us a set of addresses and then disconnect.
//
static std::chrono::seconds const kBootcacheCooldownTime(60);
//------------------------------------------------------------------------------
//
// Livecache
//
//------------------------------------------------------------------------------
// Drop incoming messages with hops greater than this number
constexpr std::uint32_t kMaxHops = 6;
// How many Endpoint to send in each mtENDPOINTS
constexpr std::uint32_t kNumberOfEndpoints = 2 * kMaxHops;
// The most Endpoint we will accept in mtENDPOINTS
constexpr std::uint32_t kNumberOfEndpointsMax =
std::max<decltype(kNumberOfEndpoints)>(kNumberOfEndpoints * 2, 64);
// Number of addresses we provide when redirecting.
constexpr std::uint32_t kRedirectEndpointCount = 10;
// How often we send or accept mtENDPOINTS messages per peer
// (we use a prime number of purpose)
constexpr std::chrono::seconds kSecondsPerMessage(151);
// How long an Endpoint will stay in the cache
// This should be a small multiple of the broadcast frequency
constexpr std::chrono::seconds kLiveCacheSecondsToLive(30);
// How much time to wait before trying an outgoing address again.
// Note that we ignore the port for purposes of comparison.
constexpr std::chrono::seconds kRecentAttemptDuration(60);
} // namespace xrpl::PeerFinder::Tuning
/** @} */

View File

@@ -0,0 +1,36 @@
#pragma once
#include <xrpl/beast/insight/Collector.h>
#include <xrpl/beast/utility/Journal.h>
#include <xrpl/peerfinder/PeerfinderManager.h>
#include <xrpl/peerfinder/Types.h>
#include <xrpl/peerfinder/detail/Store.h>
#include <boost/asio/io_context.hpp>
#include <memory>
namespace xrpl::PeerFinder {
/**
* @brief Create a new Manager.
*
* @param ioContext The io_context used to schedule asynchronous work.
* @param clock The clock used for timekeeping.
* @param journal The journal used for logging.
* @param store The persistence backend for the bootstrap cache. The caller
* retains ownership and must keep it alive (and opened) for the lifetime of
* the returned Manager. This lets consumers supply their own Store
* implementation (e.g. the SQLite-backed StoreSqdb in xrpld).
* @param collector The collector used to report metrics.
* @return The newly created Manager.
*/
std::unique_ptr<Manager>
makeManager(
boost::asio::io_context& ioContext,
clock_type& clock,
beast::Journal journal,
Store& store,
beast::insight::Collector::ptr const& collector);
} // namespace xrpl::PeerFinder

View File

@@ -15,6 +15,7 @@
// Add new amendments to the top of this list.
// Keep it sorted in reverse chronological order.
XRPL_FIX (Cleanup3_4_0, Supported::Yes, VoteBehavior::DefaultNo)
XRPL_FEATURE(Sponsor, Supported::Yes, VoteBehavior::DefaultNo)
XRPL_FEATURE(BatchV1_1, Supported::Yes, VoteBehavior::DefaultNo)
XRPL_FEATURE(LendingProtocolV1_1, Supported::No, VoteBehavior::DefaultNo)

View File

@@ -26,10 +26,10 @@ public:
explicit Family() = default;
virtual ~Family() = default;
virtual NodeStore::Database&
virtual node_store::Database&
db() = 0;
[[nodiscard]] virtual NodeStore::Database const&
[[nodiscard]] virtual node_store::Database const&
db() const = 0;
virtual beast::Journal const&

View File

@@ -17,7 +17,8 @@ class ValidPermissionedDEX
bool regularOffers_ = false; // post-fixCleanup3_2_0: excludes deleted offers
bool badHybridsOld_ = false; // pre-fixCleanup3_1_3: missing field/domain or size > 1
bool badHybrids_ = false; // post-fixCleanup3_1_3: also catches size == 0 (size != 1)
hash_set<uint256> domains_;
hash_set<uint256> domainsOld_; // pre-fixCleanup3_4_0: also flags deleted domains
hash_set<uint256> domains_; // post-fixCleanup3_4_0: excludes deleted domains
public:
void