mirror of
https://github.com/XRPLF/rippled.git
synced 2026-10-01 09:18:04 +00:00
ec0bfe521d
MetricsRegistry did two jobs. It owned the OTel metrics pipeline, and it registered the observable gauges whose callbacks read live application services. The second job is what made the whole class xrpld-tier, so the pipeline's lifecycle -- the recording() gate and the stop() teardown that closes a use-after-free window -- could not be unit-tested in xrpl_tests. Split it in two: - xrpl::telemetry::MetricsRegistry (libxrpl) owns the exporter, provider, meter, the 16 synchronous instruments, recording(), stop(), and the record*/increment* methods. - xrpl::telemetry::AppMetricGauges (xrpld) owns the 19 observable gauges and their callbacks, holding a reference to the core and to the ServiceRegistry. MetricMacros.h and ValidationTracker move with the core. The macros need only recording() and meter(), both core members; the core holds a tracker by value, and a libxrpl header cannot include one from src/. ApplicationImp owns both objects and sequences them. The core is built in the member-init list, so every synchronous instrument exists before any subsystem can record one. The gauges are armed once overlay_ exists, the last service their callbacks read. Shutdown detaches the gauge callbacks before the core drops the provider, and each shutdown step is isolated so a failure in one cannot skip the others. That detach call is new. detachCallbacks() had no callers, and the flag it sets is read by the gauge callbacks but can no longer be written by the core, so the caller now has to make the ordering explicit. The telemetry module links xrpl.libxrpl.core and xrpl.libxrpl.protocol PUBLIC: ValidationTracker.h takes a LedgerIndex and MetricMacros.h takes a ServiceRegistry, both in interfaces a consumer compiles against. Adds a MetricsRegistry gtest that drives an enabled core with telemetry on and pins the recording() gate, stop() leaving the registry inert, and stop() being idempotent. The libxrpl test tree no longer depends on xrpld.telemetry at all, and the two CMake workarounds that compiled xrpld sources into xrpl_tests are gone. Documentation and dashboard source links follow the code to their new paths, split between the two classes by which one now defines each metric.
1532 lines
54 KiB
C++
1532 lines
54 KiB
C++
#include <xrpld/app/consensus/RCLConsensus.h>
|
|
|
|
#include <xrpld/app/consensus/RCLCxLedger.h>
|
|
#include <xrpld/app/consensus/RCLCxPeerPos.h>
|
|
#include <xrpld/app/consensus/RCLCxTx.h>
|
|
#include <xrpld/app/consensus/RCLValidations.h>
|
|
#include <xrpld/app/ledger/BuildLedger.h>
|
|
#include <xrpld/app/ledger/InboundLedger.h>
|
|
#include <xrpld/app/ledger/InboundLedgers.h>
|
|
#include <xrpld/app/ledger/InboundTransactions.h>
|
|
#include <xrpld/app/ledger/LedgerMaster.h>
|
|
#include <xrpld/app/ledger/LocalTxs.h>
|
|
#include <xrpld/app/ledger/OpenLedger.h>
|
|
#include <xrpld/app/misc/FeeVote.h>
|
|
#include <xrpld/app/misc/NegativeUNLVote.h>
|
|
#include <xrpld/app/misc/TxQ.h>
|
|
#include <xrpld/app/misc/ValidatorKeys.h>
|
|
#include <xrpld/app/misc/ValidatorList.h>
|
|
#include <xrpld/overlay/Overlay.h>
|
|
#include <xrpld/overlay/predicates.h>
|
|
#include <xrpld/telemetry/PropagationHelpers.h>
|
|
|
|
#include <xrpl/basics/Log.h>
|
|
#include <xrpl/basics/Slice.h>
|
|
#include <xrpl/basics/UnorderedContainers.h>
|
|
#include <xrpl/basics/base_uint.h>
|
|
#include <xrpl/basics/chrono.h>
|
|
#include <xrpl/basics/contract.h>
|
|
#include <xrpl/basics/random.h>
|
|
#include <xrpl/beast/utility/Journal.h>
|
|
#include <xrpl/beast/utility/Zero.h>
|
|
#include <xrpl/beast/utility/instrumentation.h>
|
|
#include <xrpl/consensus/CensorshipDetector.h>
|
|
#include <xrpl/consensus/Consensus.h>
|
|
#include <xrpl/consensus/ConsensusSpanNames.h>
|
|
#include <xrpl/consensus/ConsensusTypes.h>
|
|
#include <xrpl/core/HashRouter.h>
|
|
#include <xrpl/core/Job.h>
|
|
#include <xrpl/crypto/csprng.h>
|
|
#include <xrpl/json/json_value.h>
|
|
#include <xrpl/json/json_writer.h>
|
|
#include <xrpl/ledger/AmendmentTable.h>
|
|
#include <xrpl/ledger/ApplyView.h>
|
|
#include <xrpl/ledger/Ledger.h>
|
|
#include <xrpl/ledger/LedgerTiming.h>
|
|
#include <xrpl/ledger/OpenView.h>
|
|
#include <xrpl/ledger/ReadView.h>
|
|
#include <xrpl/ledger/View.h>
|
|
#include <xrpl/protocol/BuildInfo.h>
|
|
#include <xrpl/protocol/Protocol.h>
|
|
#include <xrpl/protocol/PublicKey.h>
|
|
#include <xrpl/protocol/RippleLedgerHash.h>
|
|
#include <xrpl/protocol/Rules.h>
|
|
#include <xrpl/protocol/SField.h>
|
|
#include <xrpl/protocol/STTx.h>
|
|
#include <xrpl/protocol/STValidation.h>
|
|
#include <xrpl/protocol/SecretKey.h>
|
|
#include <xrpl/protocol/Serializer.h>
|
|
#include <xrpl/protocol/UintTypes.h>
|
|
#include <xrpl/protocol/digest.h>
|
|
#include <xrpl/protocol/tokens.h>
|
|
#include <xrpl/server/LoadFeeTrack.h>
|
|
#include <xrpl/server/NetworkOPs.h>
|
|
#include <xrpl/shamap/SHAMapItem.h>
|
|
#include <xrpl/shamap/SHAMapMissingNode.h>
|
|
#include <xrpl/shamap/SHAMapTreeNode.h>
|
|
#include <xrpl/telemetry/MetricMacros.h>
|
|
#include <xrpl/telemetry/MetricsRegistry.h>
|
|
#include <xrpl/telemetry/SpanGuard.h>
|
|
|
|
#include <boost/smart_ptr/intrusive_ptr.hpp>
|
|
|
|
#include <xrpl.pb.h>
|
|
|
|
#include <algorithm>
|
|
#include <chrono>
|
|
#include <cstddef>
|
|
#include <cstdint>
|
|
#include <exception>
|
|
#include <iomanip>
|
|
#include <limits>
|
|
#include <memory>
|
|
#include <mutex>
|
|
#include <optional>
|
|
#include <set>
|
|
#include <sstream>
|
|
#include <string>
|
|
#include <string_view>
|
|
#include <type_traits>
|
|
#include <utility>
|
|
#include <vector>
|
|
|
|
#ifdef XRPL_ENABLE_TELEMETRY
|
|
// The Telemetry interface and the shared segment names are named only by
|
|
// startRoundTracing(), which is telemetry-enabled code.
|
|
#include <xrpl/telemetry/SpanNames.h>
|
|
#include <xrpl/telemetry/Telemetry.h>
|
|
#endif
|
|
|
|
namespace xrpl {
|
|
|
|
RCLConsensus::RCLConsensus(
|
|
Application& app,
|
|
std::unique_ptr<FeeVote>&& feeVote,
|
|
LedgerMaster& ledgerMaster,
|
|
LocalTxs& localTxs,
|
|
InboundTransactions& inboundTransactions,
|
|
Consensus<Adaptor>::clock_type const& clock,
|
|
ValidatorKeys const& validatorKeys,
|
|
beast::Journal journal)
|
|
: adaptor_(
|
|
app,
|
|
std::move(feeVote),
|
|
ledgerMaster,
|
|
localTxs,
|
|
inboundTransactions,
|
|
validatorKeys,
|
|
journal)
|
|
, consensus_(clock, adaptor_, journal)
|
|
, j_(journal)
|
|
{
|
|
}
|
|
|
|
RCLConsensus::Adaptor::Adaptor(
|
|
Application& app,
|
|
std::unique_ptr<FeeVote>&& feeVote,
|
|
LedgerMaster& ledgerMaster,
|
|
LocalTxs& localTxs,
|
|
InboundTransactions& inboundTransactions,
|
|
ValidatorKeys const& validatorKeys,
|
|
beast::Journal journal)
|
|
: app_(app)
|
|
, feeVote_(std::move(feeVote))
|
|
, ledgerMaster_(ledgerMaster)
|
|
, localTxs_(localTxs)
|
|
, inboundTransactions_{inboundTransactions}
|
|
, j_(journal)
|
|
, validatorKeys_(validatorKeys)
|
|
, valCookie_(1 + randInt(cryptoPrng(), std::numeric_limits<std::uint64_t>::max() - 1))
|
|
, nUnlVote_(validatorKeys_.nodeID, j_)
|
|
{
|
|
XRPL_ASSERT(valCookie_, "xrpl::RCLConsensus::Adaptor::Adaptor : nonzero cookie");
|
|
|
|
JLOG(j_.info()) << "Consensus engine started (cookie: " + std::to_string(valCookie_) + ")";
|
|
|
|
if (validatorKeys_.nodeID != beast::kZero && validatorKeys_.keys)
|
|
{
|
|
JLOG(j_.info()) << "Validator identity: "
|
|
<< toBase58(TokenType::NodePublic, validatorKeys_.keys->masterPublicKey);
|
|
|
|
if (validatorKeys_.keys->masterPublicKey != validatorKeys_.keys->publicKey)
|
|
{
|
|
JLOG(j_.debug()) << "Validator ephemeral signing key: "
|
|
<< toBase58(TokenType::NodePublic, validatorKeys_.keys->publicKey)
|
|
<< " (seq: " << std::to_string(validatorKeys_.sequence) << ")";
|
|
}
|
|
}
|
|
}
|
|
|
|
std::optional<RCLCxLedger>
|
|
RCLConsensus::Adaptor::acquireLedger(LedgerHash const& hash)
|
|
{
|
|
// we need to switch the ledger we're working from
|
|
auto built = ledgerMaster_.getLedgerByHash(hash);
|
|
if (!built)
|
|
{
|
|
if (acquiringLedger_ != hash)
|
|
{
|
|
// need to start acquiring the correct consensus LCL
|
|
JLOG(j_.warn()) << "Need consensus ledger " << hash;
|
|
|
|
// Tell the ledger acquire system that we need the consensus ledger
|
|
acquiringLedger_ = hash;
|
|
|
|
app_.getJobQueue().addJob(JtAdvance, "GetConsL1", [id = hash, &app = app_, this]() {
|
|
JLOG(j_.debug()) << "JOB advanceLedger getConsensusLedger1 started";
|
|
app.getInboundLedgers().acquireAsync(id, 0, InboundLedger::Reason::CONSENSUS);
|
|
});
|
|
}
|
|
return std::nullopt;
|
|
}
|
|
|
|
XRPL_ASSERT(
|
|
!built->open() && built->isImmutable(),
|
|
"xrpl::RCLConsensus::Adaptor::acquireLedger : valid ledger state");
|
|
XRPL_ASSERT(
|
|
built->header().hash == hash,
|
|
"xrpl::RCLConsensus::Adaptor::acquireLedger : ledger hash match");
|
|
|
|
// Notify inbound transactions of the new ledger sequence number
|
|
inboundTransactions_.newRound(built->header().seq);
|
|
|
|
return RCLCxLedger(built);
|
|
}
|
|
|
|
void
|
|
RCLConsensus::Adaptor::share(RCLCxPeerPos const& peerPos)
|
|
{
|
|
protocol::TMProposeSet prop;
|
|
|
|
auto const& proposal = peerPos.proposal();
|
|
|
|
prop.set_proposeseq(proposal.proposeSeq());
|
|
prop.set_closetime(proposal.closeTime().time_since_epoch().count());
|
|
|
|
prop.set_currenttxhash(proposal.position().begin(), proposal.position().size());
|
|
prop.set_previousledger(proposal.prevLedger().begin(), proposal.prevLedger().size());
|
|
|
|
auto const pk = peerPos.publicKey().slice();
|
|
prop.set_nodepubkey(pk.data(), pk.size());
|
|
|
|
auto const sig = peerPos.signature();
|
|
prop.set_signature(sig.data(), sig.size());
|
|
|
|
app_.getOverlay().relay(prop, peerPos.suppressionID(), peerPos.publicKey());
|
|
}
|
|
|
|
void
|
|
RCLConsensus::Adaptor::share(RCLCxTx const& tx)
|
|
{
|
|
// If we didn't relay this transaction recently, relay it to all peers
|
|
if (app_.getHashRouter().shouldRelay(tx.id()))
|
|
{
|
|
JLOG(j_.debug()) << "Relaying disputed tx " << tx.id();
|
|
auto const slice = tx.tx->slice();
|
|
protocol::TMTransaction msg;
|
|
msg.set_rawtransaction(slice.data(), slice.size());
|
|
msg.set_status(protocol::tsNEW);
|
|
msg.set_receivetimestamp(app_.getTimeKeeper().now().time_since_epoch().count());
|
|
static std::set<Peer::id_t> const kSkip{};
|
|
app_.getOverlay().relay(tx.id(), msg, kSkip);
|
|
}
|
|
else
|
|
{
|
|
JLOG(j_.debug()) << "Not relaying disputed tx " << tx.id();
|
|
}
|
|
}
|
|
void
|
|
RCLConsensus::Adaptor::propose(RCLCxPeerPos::Proposal const& proposal)
|
|
{
|
|
// Child of the round span via its captured context (roundSpan_ is a
|
|
// thread-free SpanGuard, so parent explicitly via its context).
|
|
auto span = telemetry::SpanGuard::childSpan(
|
|
telemetry::consensus::span::proposalSend, roundSpanContext_);
|
|
span.setAttribute(
|
|
telemetry::consensus::span::attr::round, static_cast<int64_t>(proposal.proposeSeq()));
|
|
span.setAttribute(telemetry::consensus::span::attr::isBowOut, proposal.isBowOut());
|
|
|
|
JLOG(j_.trace()) << (proposal.isBowOut() ? "We bow out: " : "We propose: ")
|
|
<< xrpl::to_string(proposal.prevLedger()) << " -> "
|
|
<< xrpl::to_string(proposal.position());
|
|
|
|
protocol::TMProposeSet prop;
|
|
|
|
prop.set_currenttxhash(proposal.position().begin(), proposal.position().size());
|
|
prop.set_previousledger(proposal.prevLedger().begin(), proposal.prevLedger().size());
|
|
prop.set_proposeseq(proposal.proposeSeq());
|
|
prop.set_closetime(proposal.closeTime().time_since_epoch().count());
|
|
|
|
if (!validatorKeys_.keys)
|
|
{
|
|
JLOG(j_.warn()) << "RCLConsensus::Adaptor::propose: ValidatorKeys "
|
|
"not set: \n";
|
|
return;
|
|
}
|
|
|
|
auto const& keys = *validatorKeys_.keys;
|
|
|
|
prop.set_nodepubkey(keys.publicKey.data(), keys.publicKey.size());
|
|
|
|
auto sig = signDigest(keys.publicKey, keys.secretKey, proposal.signingHash());
|
|
|
|
prop.set_signature(sig.data(), sig.size());
|
|
|
|
auto const suppression = proposalUniqueId(
|
|
proposal.position(),
|
|
proposal.prevLedger(),
|
|
proposal.proposeSeq(),
|
|
proposal.closeTime(),
|
|
keys.publicKey,
|
|
sig);
|
|
|
|
app_.getHashRouter().addSuppression(suppression);
|
|
|
|
// Inject the current thread's active span context (e.g. the consensus
|
|
// round span) so receiving peers can link their proposal.receive span
|
|
// as a child of this trace.
|
|
//
|
|
// The helper injects only when a span is actually active, so a node with
|
|
// telemetry compiled out, disabled by config, or simply not tracing this
|
|
// round sends no TraceContext at all rather than an empty one that makes
|
|
// every peer take its has_trace_context() branch for nothing.
|
|
telemetry::injectCurrentContext(prop);
|
|
|
|
app_.getOverlay().broadcast(prop);
|
|
}
|
|
|
|
void
|
|
RCLConsensus::Adaptor::share(RCLTxSet const& txns)
|
|
{
|
|
inboundTransactions_.giveSet(txns.id(), txns.map, false);
|
|
}
|
|
|
|
std::optional<RCLTxSet>
|
|
RCLConsensus::Adaptor::acquireTxSet(RCLTxSet::ID const& setId)
|
|
{
|
|
if (auto txns = inboundTransactions_.getSet(setId, true))
|
|
{
|
|
return RCLTxSet{std::move(txns)};
|
|
}
|
|
return std::nullopt;
|
|
}
|
|
|
|
bool
|
|
RCLConsensus::Adaptor::hasOpenTransactions() const
|
|
{
|
|
return !app_.getOpenLedger().empty();
|
|
}
|
|
|
|
std::size_t
|
|
RCLConsensus::Adaptor::proposersValidated(LedgerHash const& h) const
|
|
{
|
|
return app_.getValidations().numTrustedForLedger(h);
|
|
}
|
|
|
|
std::size_t
|
|
RCLConsensus::Adaptor::proposersFinished(RCLCxLedger const& ledger, LedgerHash const& h) const
|
|
{
|
|
RCLValidations& vals = app_.getValidations();
|
|
return vals.getNodesAfter(RCLValidatedLedger(ledger.ledger, vals.adaptor().journal()), h);
|
|
}
|
|
|
|
uint256
|
|
RCLConsensus::Adaptor::getPrevLedger(
|
|
uint256 ledgerID,
|
|
RCLCxLedger const& ledger,
|
|
ConsensusMode mode)
|
|
{
|
|
RCLValidations& vals = app_.getValidations();
|
|
uint256 netLgr = vals.getPreferred(
|
|
RCLValidatedLedger{ledger.ledger, vals.adaptor().journal()},
|
|
ledgerMaster_.getValidLedgerIndex());
|
|
|
|
if (netLgr != ledgerID)
|
|
{
|
|
if (mode != ConsensusMode::WrongLedger)
|
|
app_.getOPs().consensusViewChange();
|
|
|
|
JLOG(j_.debug()) << json::Compact(app_.getValidations().getJsonTrie());
|
|
}
|
|
|
|
return netLgr;
|
|
}
|
|
|
|
auto
|
|
RCLConsensus::Adaptor::onClose(
|
|
RCLCxLedger const& ledger,
|
|
NetClock::time_point const& closeTime,
|
|
ConsensusMode mode) -> Result
|
|
{
|
|
namespace cs = telemetry::consensus::span;
|
|
|
|
// Child of the round span via its captured context (roundSpan_ is a
|
|
// thread-free SpanGuard, so parent explicitly via its context).
|
|
auto span = telemetry::SpanGuard::childSpan(cs::ledgerClose, roundSpanContext_);
|
|
// setAttribute is the only consumer of everything read here, so the block is
|
|
// guarded on the span being live. Unguarded, every round takes the open
|
|
// ledger's currentMutex_ and copies a shared_ptr just to read txCount, and
|
|
// builds a mode string, for attributes no one may be recording.
|
|
if (span)
|
|
{
|
|
span.setAttribute(
|
|
cs::attr::ledgerSeq, static_cast<int64_t>(ledger.ledger->header().seq) + 1);
|
|
span.setAttribute(cs::attr::mode, toDisplayString(mode).c_str());
|
|
span.setAttribute(
|
|
cs::attr::txCountOpen, static_cast<int64_t>(app_.getOpenLedger().current()->txCount()));
|
|
span.setAttribute(
|
|
cs::attr::closeTimeResolutionMs,
|
|
static_cast<int64_t>(
|
|
std::chrono::duration_cast<std::chrono::milliseconds>(ledger.closeTimeResolution())
|
|
.count()));
|
|
}
|
|
|
|
bool const wrongLCL = mode == ConsensusMode::WrongLedger;
|
|
bool const proposing = mode == ConsensusMode::Proposing;
|
|
|
|
notify(protocol::neCLOSING_LEDGER, ledger, !wrongLCL);
|
|
|
|
auto const& prevLedger = ledger.ledger;
|
|
|
|
ledgerMaster_.applyHeldTransactions();
|
|
// Tell the ledger master not to acquire the ledger we're probably building
|
|
ledgerMaster_.setBuildingLedger(prevLedger->header().seq + 1);
|
|
|
|
auto initialLedger = app_.getOpenLedger().current();
|
|
|
|
auto initialSet = std::make_shared<SHAMap>(SHAMapType::TRANSACTION, app_.getNodeFamily());
|
|
initialSet->setUnbacked();
|
|
|
|
// Build SHAMap containing all transactions in our open ledger
|
|
for (auto const& tx : initialLedger->txs)
|
|
{
|
|
JLOG(j_.trace()) << "Adding open ledger TX " << tx.first->getTransactionID();
|
|
Serializer s(2048);
|
|
tx.first->add(s);
|
|
initialSet->addItem(
|
|
SHAMapNodeType::TnTransactionNm,
|
|
makeShamapitem(tx.first->getTransactionID(), s.slice()));
|
|
}
|
|
|
|
// Add pseudo-transactions to the set
|
|
if (app_.config().standalone() || (proposing && !wrongLCL))
|
|
{
|
|
if (prevLedger->isFlagLedger())
|
|
{
|
|
// previous ledger was flag ledger, add fee and amendment
|
|
// pseudo-transactions
|
|
auto validations =
|
|
app_.getValidators().negativeUNLFilter(app_.getValidations().getTrustedForLedger(
|
|
prevLedger->header().parentHash, prevLedger->seq() - 1));
|
|
if (validations.size() >= app_.getValidators().quorum())
|
|
{
|
|
feeVote_->doVoting(prevLedger, validations, initialSet);
|
|
app_.getAmendmentTable().doVoting(prevLedger, validations, initialSet, j_);
|
|
}
|
|
}
|
|
else if (prevLedger->isVotingLedger())
|
|
{
|
|
// previous ledger was a voting ledger,
|
|
// so the current consensus session is for a flag ledger,
|
|
// add negative UNL pseudo-transactions
|
|
nUnlVote_.doVoting(
|
|
prevLedger,
|
|
app_.getValidators().getTrustedMasterKeys(),
|
|
app_.getValidations(),
|
|
initialSet);
|
|
}
|
|
}
|
|
|
|
// Now we need an immutable snapshot
|
|
initialSet = initialSet->snapShot(false);
|
|
|
|
if (!wrongLCL)
|
|
{
|
|
LedgerIndex const seq = prevLedger->header().seq + 1;
|
|
CensorshipDetector<TxID, LedgerIndex>::TxIDSeqVec proposed;
|
|
|
|
initialSet->visitLeaves(
|
|
[&proposed, seq](boost::intrusive_ptr<SHAMapItem const> const& item) {
|
|
proposed.emplace_back(item->key(), seq);
|
|
});
|
|
|
|
censorshipDetector_.propose(std::move(proposed));
|
|
}
|
|
|
|
// Needed because of the move below.
|
|
auto const setHash = initialSet->getHash().asUInt256();
|
|
|
|
return Result{
|
|
std::move(initialSet),
|
|
RCLCxPeerPos::Proposal{
|
|
initialLedger->header().parentHash,
|
|
RCLCxPeerPos::Proposal::kSeqJoin,
|
|
setHash,
|
|
closeTime,
|
|
app_.getTimeKeeper().closeTime(),
|
|
validatorKeys_.nodeID}};
|
|
}
|
|
|
|
void
|
|
RCLConsensus::Adaptor::onForceAccept(
|
|
Result const& result,
|
|
RCLCxLedger const& prevLedger,
|
|
NetClock::duration const& closeResolution,
|
|
ConsensusCloseTimes const& rawCloseTimes,
|
|
ConsensusMode const& mode,
|
|
json::Value&& consensusJson)
|
|
{
|
|
auto acceptSpan = makeAcceptSpan(result);
|
|
doAccept(
|
|
result,
|
|
prevLedger,
|
|
closeResolution,
|
|
rawCloseTimes,
|
|
mode,
|
|
std::move(consensusJson),
|
|
std::move(acceptSpan));
|
|
}
|
|
|
|
void
|
|
RCLConsensus::Adaptor::onAccept(
|
|
Result const& result,
|
|
RCLCxLedger const& prevLedger,
|
|
NetClock::duration const& closeResolution,
|
|
ConsensusCloseTimes const& rawCloseTimes,
|
|
ConsensusMode const& mode,
|
|
json::Value&& consensusJson,
|
|
bool const validating)
|
|
{
|
|
auto acceptSpan = makeAcceptSpan(result);
|
|
|
|
app_.getJobQueue().addJob(
|
|
JtAccept,
|
|
"AcceptLedger",
|
|
// NOLINTNEXTLINE(cppcoreguidelines-misleading-capture-default-by-value)
|
|
[=, this, cj = std::move(consensusJson), sp = std::move(acceptSpan)]() mutable {
|
|
// Note that no lock is held or acquired during this job.
|
|
// This is because generic Consensus guarantees that once a ledger
|
|
// is accepted, the consensus results and capture by reference state
|
|
// will not change until startRound is called (which happens via
|
|
// endConsensus).
|
|
RclConsensusLogger clog("onAccept", validating, j_);
|
|
this->doAccept(
|
|
result,
|
|
prevLedger,
|
|
closeResolution,
|
|
rawCloseTimes,
|
|
mode,
|
|
std::move(cj),
|
|
std::move(sp));
|
|
this->app_.getOPs().endConsensus(clog.ss());
|
|
});
|
|
}
|
|
|
|
// Not static: the guarded body reads app_ and roundSpanContext_. With telemetry
|
|
// compiled out it returns an empty handle and touches no member, so clang-tidy
|
|
// sees a method that could be static.
|
|
// NOLINTBEGIN(readability-convert-member-functions-to-static)
|
|
std::shared_ptr<telemetry::SpanGuard>
|
|
RCLConsensus::Adaptor::makeAcceptSpan(Result const& result)
|
|
{
|
|
// The whole body is telemetry: the guard, its attributes and the captured
|
|
// context serve the accept span only. With telemetry compiled out the handle
|
|
// stays empty, so accepting a ledger does not allocate a control block for a
|
|
// span that can never record. doAccept only hands the handle to
|
|
// activateIfLive(), which tests it, so an empty handle is safe on both the
|
|
// sync (onForceAccept) and async (onAccept) paths.
|
|
#ifdef XRPL_ENABLE_TELEMETRY
|
|
namespace cs = telemetry::consensus::span;
|
|
|
|
auto span = std::make_shared<telemetry::SpanGuard>(
|
|
telemetry::SpanGuard::childSpan(cs::accept, roundSpanContext_));
|
|
|
|
// Every attribute below exists only for the span, so the whole block —
|
|
// attributes and the context capture — is guarded on the span being live.
|
|
if (*span)
|
|
{
|
|
span->setAttribute(cs::attr::proposers, static_cast<int64_t>(result.proposers));
|
|
span->setAttribute(
|
|
cs::attr::roundTimeMs, static_cast<int64_t>(result.roundTime.read().count()));
|
|
span->setAttribute(cs::attr::quorum, static_cast<int64_t>(app_.getValidators().quorum()));
|
|
span->setAttribute(cs::attr::disputesCount, static_cast<int64_t>(result.disputes.size()));
|
|
char const* stateStr = [&] {
|
|
switch (result.state)
|
|
{
|
|
case ConsensusState::Yes:
|
|
return "yes";
|
|
case ConsensusState::MovedOn:
|
|
return "moved_on";
|
|
case ConsensusState::Expired:
|
|
return "expired";
|
|
default:
|
|
return "no";
|
|
}
|
|
}();
|
|
span->setAttribute(cs::attr::consensusState, stateStr);
|
|
|
|
// Capture the accept span's context so createValidationSpan() — which
|
|
// runs on the jtACCEPT worker thread — can link the validation.send
|
|
// span to the accept span (matching the design diagram and the
|
|
// "validation follows acceptance" causal model).
|
|
//
|
|
// span is a thread-free SpanGuard handed to the JtAccept worker
|
|
// (onAccept), which ends it there. spanContext() captures the guard's
|
|
// own span, so accept.apply parents via acceptSpanContext_ regardless
|
|
// of which thread runs the sync (onForceAccept) or async (onAccept)
|
|
// path -- no scope work is needed.
|
|
acceptSpanContext_ = span->spanContext();
|
|
}
|
|
return span;
|
|
#else
|
|
return {};
|
|
#endif
|
|
}
|
|
// NOLINTEND(readability-convert-member-functions-to-static)
|
|
|
|
void
|
|
RCLConsensus::Adaptor::doAccept(
|
|
Result const& result,
|
|
RCLCxLedger const& prevLedger,
|
|
NetClock::duration closeResolution,
|
|
ConsensusCloseTimes const& rawCloseTimes,
|
|
ConsensusMode const& mode,
|
|
json::Value&& consensusJson,
|
|
std::shared_ptr<telemetry::SpanGuard> acceptSpan)
|
|
{
|
|
namespace cs = telemetry::consensus::span;
|
|
|
|
// Make the accept span ambient for the whole accept so doAccept's log lines
|
|
// (and any spans created here) correlate to it. Non-owning: acceptSpan still
|
|
// owns/ends the span. doAccept runs to completion on the JtAccept worker
|
|
// (no coroutine yield), so this scope is thread-local and safe.
|
|
auto acceptActivation = telemetry::activateIfLive(acceptSpan);
|
|
|
|
prevProposers_ = result.proposers;
|
|
prevRoundTime_ = result.roundTime.read();
|
|
|
|
bool closeTimeCorrect = false;
|
|
|
|
bool const proposing = mode == ConsensusMode::Proposing;
|
|
bool const haveCorrectLCL = mode != ConsensusMode::WrongLedger;
|
|
bool const consensusFail = result.state == ConsensusState::MovedOn;
|
|
|
|
auto consensusCloseTime = result.position.closeTime();
|
|
|
|
if (consensusCloseTime == NetClock::time_point{})
|
|
{
|
|
// We agreed to disagree on the close time
|
|
using namespace std::chrono_literals;
|
|
consensusCloseTime = prevLedger.closeTime() + 1s;
|
|
closeTimeCorrect = false;
|
|
}
|
|
else
|
|
{
|
|
// We agreed on a close time
|
|
consensusCloseTime =
|
|
effCloseTime(consensusCloseTime, closeResolution, prevLedger.closeTime());
|
|
closeTimeCorrect = true;
|
|
}
|
|
|
|
// Parent accept.apply via the captured accept context (acceptSpanContext_):
|
|
// the accept span is a thread-free SpanGuard, so an explicit context is
|
|
// used for both the sync (onForceAccept) and async (onAccept) paths. Falls
|
|
// back to the round context if the accept span was null.
|
|
auto doAcceptSpan = acceptSpanContext_.isValid()
|
|
? telemetry::SpanGuard::childSpan(cs::acceptApply, acceptSpanContext_)
|
|
: telemetry::SpanGuard::childSpan(cs::acceptApply, roundSpanContext_);
|
|
doAcceptSpan.setAttribute(cs::attr::ledgerSeq, static_cast<int64_t>(prevLedger.seq()) + 1);
|
|
doAcceptSpan.setAttribute(
|
|
cs::attr::closeTimeRippleEpochS,
|
|
static_cast<int64_t>(consensusCloseTime.time_since_epoch().count()));
|
|
doAcceptSpan.setAttribute(cs::attr::closeTimeCorrect, closeTimeCorrect);
|
|
doAcceptSpan.setAttribute(
|
|
cs::attr::closeResolutionMs,
|
|
static_cast<int64_t>(
|
|
std::chrono::duration_cast<std::chrono::milliseconds>(closeResolution).count()));
|
|
doAcceptSpan.setAttribute(
|
|
cs::attr::consensusState,
|
|
consensusFail ? std::string_view{cs::val::movedOn} : std::string_view{cs::val::finished});
|
|
doAcceptSpan.setAttribute(cs::attr::proposing, proposing);
|
|
doAcceptSpan.setAttribute(
|
|
cs::attr::roundTimeMs, static_cast<int64_t>(result.roundTime.read().count()));
|
|
doAcceptSpan.setAttribute(
|
|
cs::attr::parentCloseTimeRippleEpochS,
|
|
static_cast<int64_t>(prevLedger.closeTime().time_since_epoch().count()));
|
|
doAcceptSpan.setAttribute(
|
|
cs::attr::closeTimeSelfRippleEpochS,
|
|
static_cast<int64_t>(rawCloseTimes.self.time_since_epoch().count()));
|
|
doAcceptSpan.setAttribute(
|
|
cs::attr::closeTimeVoteBins, static_cast<int64_t>(rawCloseTimes.peers.size()));
|
|
doAcceptSpan.setAttribute(
|
|
cs::attr::disputesResolvedCount, static_cast<int64_t>(result.disputes.size()));
|
|
// prevRes and dir feed the resolution_direction attribute and nothing else,
|
|
// so both are guarded on the span being active. Unguarded, every accepted
|
|
// ledger builds a std::string that no one reads.
|
|
if (doAcceptSpan)
|
|
{
|
|
auto const prevRes = prevLedger.closeTimeResolution();
|
|
auto const dir = [&]() -> std::string {
|
|
if (closeResolution > prevRes)
|
|
{
|
|
return "increased";
|
|
}
|
|
if (closeResolution < prevRes)
|
|
{
|
|
return "decreased";
|
|
}
|
|
return "unchanged";
|
|
}();
|
|
doAcceptSpan.setAttribute(cs::attr::resolutionDirection, std::move(dir));
|
|
}
|
|
|
|
JLOG(j_.debug()) << "Report: Prop=" << (proposing ? "yes" : "no")
|
|
<< " val=" << (validating_ ? "yes" : "no")
|
|
<< " corLCL=" << (haveCorrectLCL ? "yes" : "no")
|
|
<< " fail=" << (consensusFail ? "yes" : "no");
|
|
JLOG(j_.debug()) << "Report: Prev = " << prevLedger.id() << ":" << prevLedger.seq();
|
|
|
|
//--------------------------------------------------------------------------
|
|
std::set<TxID> failed;
|
|
|
|
// We want to put transactions in an unpredictable but deterministic order:
|
|
// we use the hash of the set.
|
|
//
|
|
// FIXME: Use a std::vector and a custom sorter instead of CanonicalTXSet?
|
|
CanonicalTXSet retriableTxs{result.txns.map->getHash().asUInt256()};
|
|
|
|
JLOG(j_.debug()) << "Building canonical tx set: " << retriableTxs.key();
|
|
|
|
// One tx.included event per transaction of the agreed consensus set, which
|
|
// is not yet the accepted ledger: buildLCL() below applies these and some
|
|
// may fail, so the events are a superset of what the ledger ends up with. A
|
|
// transaction whose bytes cannot be parsed gets no event at all.
|
|
//
|
|
// txCount and the per-transaction event feed the span and nothing else, so
|
|
// both are guarded on the span being active. Unguarded, every accepted
|
|
// ledger builds one 64-character hash string per transaction that no one
|
|
// reads.
|
|
int64_t txCount = 0;
|
|
for (auto const& item : *result.txns.map)
|
|
{
|
|
try
|
|
{
|
|
retriableTxs.insert(std::make_shared<STTx const>(SerialIter{item.slice()}));
|
|
JLOG(j_.debug()) << " Tx: " << item.key();
|
|
if (doAcceptSpan)
|
|
{
|
|
++txCount;
|
|
auto const txHash = to_string(item.key());
|
|
doAcceptSpan.addEvent(cs::event::txIncluded, {{cs::attr::txId, txHash}});
|
|
}
|
|
}
|
|
catch (std::exception const& ex)
|
|
{
|
|
failed.insert(item.key());
|
|
JLOG(j_.warn()) << " Tx: " << item.key() << " throws: " << ex.what();
|
|
}
|
|
}
|
|
if (doAcceptSpan)
|
|
{
|
|
doAcceptSpan.setAttribute(cs::attr::txCount, txCount);
|
|
}
|
|
|
|
auto built = buildLCL(
|
|
prevLedger,
|
|
retriableTxs,
|
|
consensusCloseTime,
|
|
closeTimeCorrect,
|
|
closeResolution,
|
|
result.roundTime.read(),
|
|
failed);
|
|
|
|
auto const newLCLHash = built.id();
|
|
JLOG(j_.debug()) << "Built ledger #" << built.seq() << ": " << newLCLHash;
|
|
|
|
// Tell directly connected peers that we have a new LCL
|
|
notify(protocol::neACCEPTED_LEDGER, built, haveCorrectLCL);
|
|
|
|
// As long as we're in sync with the network, attempt to detect attempts
|
|
// at censorship of transaction by tracking which ones don't make it in
|
|
// after a period of time.
|
|
if (haveCorrectLCL && result.state == ConsensusState::Yes)
|
|
{
|
|
std::vector<TxID> accepted;
|
|
|
|
result.txns.map->visitLeaves(
|
|
[&accepted](boost::intrusive_ptr<SHAMapItem const> const& item) {
|
|
accepted.push_back(item->key());
|
|
});
|
|
|
|
// Track all the transactions which failed or were marked as retriable
|
|
for (auto const& r : retriableTxs)
|
|
failed.insert(r.first.getTXID());
|
|
|
|
censorshipDetector_.check(
|
|
std::move(accepted),
|
|
[curr = built.seq(), j = app_.getJournal("CensorshipDetector"), &failed](
|
|
uint256 const& id, LedgerIndex seq) {
|
|
if (failed.contains(id))
|
|
return true;
|
|
|
|
auto const wait = curr - seq;
|
|
|
|
if (wait && (wait % kCensorshipWarnInternal == 0))
|
|
{
|
|
std::ostringstream ss;
|
|
ss << "Potential Censorship: Eligible tx " << id
|
|
<< ", which we are tracking since ledger " << seq
|
|
<< " has not been included as of ledger " << curr << ".";
|
|
|
|
JLOG(j.warn()) << ss.str();
|
|
}
|
|
|
|
return false;
|
|
});
|
|
}
|
|
|
|
if (validating_)
|
|
validating_ = ledgerMaster_.isCompatible(*built.ledger, j_.warn(), "Not validating");
|
|
|
|
if (validating_ && !consensusFail && app_.getValidations().canValidateSeq(built.seq()))
|
|
{
|
|
validate(built, result.txns, proposing);
|
|
JLOG(j_.info()) << "CNF Val " << newLCLHash;
|
|
}
|
|
else
|
|
{
|
|
JLOG(j_.info()) << "CNF buildLCL " << newLCLHash;
|
|
}
|
|
|
|
// See if we can accept a ledger as fully-validated
|
|
ledgerMaster_.consensusBuilt(built.ledger, result.txns.id(), std::move(consensusJson));
|
|
|
|
// Record ledger close for OTel dashboard parity counter. Uses the
|
|
// call-site macro (see MetricMacros.h) rather than a MetricsRegistry
|
|
// member.
|
|
XRPL_METRIC_COUNTER_INC(app_, "ledgers_closed_total", "Total ledgers closed by consensus");
|
|
|
|
//-------------------------------------------------------------------------
|
|
{
|
|
// Apply disputed transactions that didn't get in
|
|
//
|
|
// The first crack of transactions to get into the new
|
|
// open ledger goes to transactions proposed by a validator
|
|
// we trust but not included in the consensus set.
|
|
//
|
|
// These are done first because they are the most likely
|
|
// to receive agreement during consensus. They are also
|
|
// ordered logically "sooner" than transactions not mentioned
|
|
// in the previous consensus round.
|
|
//
|
|
bool anyDisputes = false;
|
|
for (auto const& [_, dispute] : result.disputes)
|
|
{
|
|
(void)_;
|
|
if (!dispute.getOurVote())
|
|
{
|
|
// we voted NO
|
|
try
|
|
{
|
|
JLOG(j_.debug()) << "Test applying disputed transaction that did"
|
|
<< " not get in " << dispute.tx().id();
|
|
|
|
SerialIter sit(dispute.tx().tx->slice());
|
|
auto txn = std::make_shared<STTx const>(sit);
|
|
|
|
// Disputed pseudo-transactions that were not accepted
|
|
// can't be successfully applied in the next ledger
|
|
if (isPseudoTx(*txn))
|
|
continue;
|
|
|
|
retriableTxs.insert(txn);
|
|
|
|
anyDisputes = true;
|
|
}
|
|
catch (std::exception const& ex)
|
|
{
|
|
JLOG(j_.debug()) << "Failed to apply transaction we voted "
|
|
"NO on. Exception: "
|
|
<< ex.what();
|
|
}
|
|
}
|
|
}
|
|
|
|
// Build new open ledger
|
|
std::unique_lock lock{app_.getMasterMutex(), std::defer_lock};
|
|
std::unique_lock sl{ledgerMaster_.peekMutex(), std::defer_lock};
|
|
std::lock(lock, sl);
|
|
|
|
auto const lastVal = ledgerMaster_.getValidatedLedger();
|
|
std::optional<Rules> rules;
|
|
if (lastVal)
|
|
{
|
|
rules = makeRulesGivenLedger(*lastVal, app_.config().features);
|
|
}
|
|
else
|
|
{
|
|
rules.emplace(app_.config().features);
|
|
}
|
|
app_.getOpenLedger().accept(
|
|
app_,
|
|
*rules,
|
|
built.ledger,
|
|
localTxs_.getTxSet(),
|
|
anyDisputes,
|
|
retriableTxs,
|
|
TapNone,
|
|
"consensus",
|
|
[&](OpenView& view, beast::Journal j) {
|
|
// Stuff the ledger with transactions from the queue.
|
|
return app_.getTxQ().accept(app_, view);
|
|
});
|
|
|
|
// Signal a potential fee change to subscribers after the open ledger
|
|
// is created
|
|
app_.getOPs().reportFeeChange();
|
|
}
|
|
|
|
//-------------------------------------------------------------------------
|
|
{
|
|
ledgerMaster_.switchLCL(built.ledger);
|
|
|
|
// Do these need to exist?
|
|
XRPL_ASSERT(
|
|
ledgerMaster_.getClosedLedger()->header().hash == built.id(),
|
|
"xrpl::RCLConsensus::Adaptor::doAccept : ledger hash match");
|
|
XRPL_ASSERT(
|
|
app_.getOpenLedger().current()->header().parentHash == built.id(),
|
|
"xrpl::RCLConsensus::Adaptor::doAccept : parent hash match");
|
|
}
|
|
|
|
//-------------------------------------------------------------------------
|
|
// we entered the round with the network,
|
|
// see how close our close time is to other node's
|
|
// close time reports, and update our clock.
|
|
if ((mode == ConsensusMode::Proposing || mode == ConsensusMode::Observing) && !consensusFail)
|
|
{
|
|
JLOG(j_.info()) << "We closed at " << rawCloseTimes.self.time_since_epoch().count();
|
|
int closeCount = 1;
|
|
for (auto const& [t, v] : rawCloseTimes.peers)
|
|
{
|
|
JLOG(j_.info()) << std::to_string(v) << " time votes for "
|
|
<< std::to_string(t.time_since_epoch().count());
|
|
closeCount += v;
|
|
}
|
|
|
|
// Median handles outliers better than mean.
|
|
auto const offset = medianCloseOffset(rawCloseTimes);
|
|
JLOG(j_.info()) << "Our close offset is estimated at " << offset.count() << " ("
|
|
<< closeCount << ")";
|
|
|
|
app_.getTimeKeeper().adjustCloseTime(offset);
|
|
}
|
|
}
|
|
|
|
void
|
|
RCLConsensus::Adaptor::notify(
|
|
protocol::NodeEvent ne,
|
|
RCLCxLedger const& ledger,
|
|
bool haveCorrectLCL)
|
|
{
|
|
protocol::TMStatusChange s;
|
|
|
|
if (!haveCorrectLCL)
|
|
{
|
|
s.set_newevent(protocol::neLOST_SYNC);
|
|
}
|
|
else
|
|
{
|
|
s.set_newevent(ne);
|
|
}
|
|
|
|
s.set_ledgerseq(ledger.seq());
|
|
s.set_networktime(app_.getTimeKeeper().now().time_since_epoch().count());
|
|
s.set_ledgerhashprevious(
|
|
ledger.parentID().begin(), std::decay_t<decltype(ledger.parentID())>::kBytes);
|
|
s.set_ledgerhash(ledger.id().begin(), std::decay_t<decltype(ledger.id())>::kBytes);
|
|
|
|
std::uint32_t uMin = 0, uMax = 0;
|
|
if (!ledgerMaster_.getFullValidatedRange(uMin, uMax))
|
|
{
|
|
uMin = 0;
|
|
uMax = 0;
|
|
}
|
|
else
|
|
{
|
|
// Don't advertise ledgers we're not willing to serve
|
|
uMin = std::max(uMin, ledgerMaster_.getEarliestFetch());
|
|
}
|
|
s.set_firstseq(uMin);
|
|
s.set_lastseq(uMax);
|
|
app_.getOverlay().foreach(SendAlways(std::make_shared<Message>(s, protocol::mtSTATUS_CHANGE)));
|
|
JLOG(j_.trace()) << "send status change to peer";
|
|
}
|
|
|
|
RCLCxLedger
|
|
RCLConsensus::Adaptor::buildLCL(
|
|
RCLCxLedger const& previousLedger,
|
|
CanonicalTXSet& retriableTxs,
|
|
NetClock::time_point closeTime,
|
|
bool closeTimeCorrect,
|
|
NetClock::duration closeResolution,
|
|
std::chrono::milliseconds roundTime,
|
|
std::set<TxID>& failedTxs)
|
|
{
|
|
std::shared_ptr<Ledger> built = [&]() {
|
|
if (auto const replayData = ledgerMaster_.releaseReplay())
|
|
{
|
|
XRPL_ASSERT(
|
|
replayData->parent()->header().hash == previousLedger.id(),
|
|
"xrpl::RCLConsensus::Adaptor::buildLCL : parent hash match");
|
|
return buildLedger(*replayData, TapNone, app_, j_);
|
|
}
|
|
return buildLedger(
|
|
previousLedger.ledger,
|
|
closeTime,
|
|
closeTimeCorrect,
|
|
closeResolution,
|
|
app_,
|
|
retriableTxs,
|
|
failedTxs,
|
|
j_);
|
|
}();
|
|
|
|
// Update fee computations based on accepted txs
|
|
using namespace std::chrono_literals;
|
|
app_.getTxQ().processClosedLedger(app_, *built, roundTime > 5s);
|
|
|
|
// And stash the ledger in the ledger master
|
|
if (ledgerMaster_.storeLedger(built))
|
|
{
|
|
JLOG(j_.debug()) << "Consensus built ledger we already had";
|
|
}
|
|
else if (app_.getInboundLedgers().find(built->header().hash))
|
|
{
|
|
JLOG(j_.debug()) << "Consensus built ledger we were acquiring";
|
|
}
|
|
else
|
|
{
|
|
JLOG(j_.debug()) << "Consensus built new ledger";
|
|
}
|
|
return RCLCxLedger{std::move(built)};
|
|
}
|
|
|
|
void
|
|
RCLConsensus::Adaptor::validate(RCLCxLedger const& ledger, RCLTxSet const& txns, bool proposing)
|
|
{
|
|
auto valSpan = createValidationSpan();
|
|
// Testing the guard as well as the optional matters: a guard that exists but
|
|
// is not live still evaluates its arguments, and the ledger_hash attribute
|
|
// below turns a 32-byte hash into a 64-character string.
|
|
if (valSpan && *valSpan)
|
|
{
|
|
namespace cs = telemetry::consensus::span;
|
|
valSpan->setAttribute(cs::attr::ledgerSeq, static_cast<int64_t>(ledger.seq()));
|
|
valSpan->setAttribute(
|
|
cs::attr::ledgerHash, to_string(ledger.ledger->header().hash).c_str());
|
|
valSpan->setAttribute(cs::attr::proposing, proposing);
|
|
// proposing implies a full validation (vfFullValidation is set on
|
|
// the STValidation only when proposing — see below).
|
|
valSpan->setAttribute(cs::attr::fullValidation, proposing);
|
|
}
|
|
|
|
using namespace std::chrono_literals;
|
|
|
|
auto validationTime = app_.getTimeKeeper().closeTime();
|
|
if (validationTime <= lastValidationTime_)
|
|
validationTime = lastValidationTime_ + 1s;
|
|
lastValidationTime_ = validationTime;
|
|
|
|
if (valSpan && *valSpan)
|
|
{
|
|
valSpan->setAttribute(
|
|
telemetry::consensus::span::attr::validationSignTime,
|
|
static_cast<int64_t>(validationTime.time_since_epoch().count()));
|
|
}
|
|
|
|
if (!validatorKeys_.keys)
|
|
{
|
|
JLOG(j_.warn()) << "RCLConsensus::Adaptor::validate: ValidatorKeys "
|
|
"not set\n";
|
|
return;
|
|
}
|
|
|
|
auto const& keys = *validatorKeys_.keys;
|
|
|
|
auto v = std::make_shared<STValidation>(
|
|
lastValidationTime_,
|
|
keys.publicKey,
|
|
keys.secretKey,
|
|
validatorKeys_.nodeID,
|
|
[&](STValidation& v) {
|
|
v.setFieldH256(sfLedgerHash, ledger.id());
|
|
v.setFieldH256(sfConsensusHash, txns.id());
|
|
|
|
v.setFieldU32(sfLedgerSequence, ledger.seq());
|
|
|
|
if (proposing)
|
|
v.setFlag(kVfFullValidation);
|
|
|
|
// Attest to the hash of what we consider to be the last fully
|
|
// validated ledger. This may be the hash of the ledger we are
|
|
// validating here, and that's fine.
|
|
if (auto const vl = ledgerMaster_.getValidatedLedger())
|
|
v.setFieldH256(sfValidatedHash, vl->header().hash);
|
|
|
|
v.setFieldU64(sfCookie, valCookie_);
|
|
|
|
// Report our server version every flag ledger:
|
|
if (ledger.ledger->isVotingLedger())
|
|
v.setFieldU64(sfServerVersion, build_info::getEncodedVersion());
|
|
|
|
// Report our load
|
|
{
|
|
auto const& ft = app_.getFeeTrack();
|
|
auto const fee = std::max(ft.getLocalFee(), ft.getClusterFee());
|
|
if (fee > ft.getLoadBase())
|
|
v.setFieldU32(sfLoadFee, fee);
|
|
}
|
|
|
|
// If the next ledger is a flag ledger, suggest fee changes and
|
|
// new features:
|
|
if (ledger.ledger->isVotingLedger())
|
|
{
|
|
// Fees:
|
|
feeVote_->doValidation(ledger.ledger->fees(), ledger.ledger->rules(), v);
|
|
|
|
// Amendments
|
|
// FIXME: pass `v` and have the function insert the array
|
|
// directly?
|
|
auto const amendments =
|
|
app_.getAmendmentTable().doValidation(getEnabledAmendments(*ledger.ledger));
|
|
|
|
if (!amendments.empty())
|
|
v.setFieldV256(sfAmendments, STVector256(sfAmendments, amendments));
|
|
}
|
|
});
|
|
|
|
auto const serialized = v->getSerialized();
|
|
|
|
// suppress it if we receive it
|
|
app_.getHashRouter().addSuppression(sha512Half(makeSlice(serialized)));
|
|
|
|
handleNewValidation(app_, v, "local");
|
|
|
|
// Broadcast to all our peers:
|
|
protocol::TMValidation val;
|
|
val.set_validation(serialized.data(), serialized.size());
|
|
// Inject the current thread's active span context so receiving
|
|
// peers can link their validation.receive span as a child.
|
|
//
|
|
// The trace_context appended below is outside the signature on
|
|
// `serialized`, so it is not covered by validation authenticity.
|
|
// Downstream consumers treat it as advisory only. A signature-covered
|
|
// trace context is a possible future enhancement.
|
|
//
|
|
// As on the proposal path, the helper injects only when a span is actually
|
|
// active, so a node that is not tracing sends no TraceContext at all
|
|
// rather than an empty one.
|
|
telemetry::injectCurrentContext(val);
|
|
app_.getOverlay().broadcast(val);
|
|
|
|
// Publish to all our subscribers:
|
|
app_.getOPs().pubValidation(v);
|
|
|
|
// Record validation sent for OTel dashboard parity counter.
|
|
if (auto* mr = app_.getMetricsRegistry())
|
|
{
|
|
mr->incrementValidationsSent();
|
|
#ifdef XRPL_ENABLE_TELEMETRY
|
|
// Record our validation for the agreement tracker so it can
|
|
// compare against network-validated ledgers.
|
|
//
|
|
// Only when enabled: recording takes the tracker's lock and inserts an
|
|
// entry, and nothing reconciles or drains those entries unless the
|
|
// observable gauges are running.
|
|
if (mr->isEnabled())
|
|
mr->getValidationTracker().recordOurValidation(ledger.id(), ledger.seq());
|
|
#endif
|
|
}
|
|
}
|
|
|
|
void
|
|
RCLConsensus::Adaptor::onModeChange(ConsensusMode before, ConsensusMode after)
|
|
{
|
|
namespace cs = telemetry::consensus::span;
|
|
|
|
// Child of the round span via its captured context (roundSpan_ is a
|
|
// thread-free SpanGuard, so parent explicitly via its context). A mode
|
|
// change outside a round leaves roundSpanContext_ invalid, yielding a null
|
|
// guard (no-op).
|
|
auto span = telemetry::SpanGuard::childSpan(cs::modeChange, roundSpanContext_);
|
|
span.setAttribute(cs::attr::modeOld, toDisplayString(before).c_str());
|
|
span.setAttribute(cs::attr::modeNew, toDisplayString(after).c_str());
|
|
|
|
JLOG(j_.info()) << "Consensus mode change before=" << to_string(before)
|
|
<< ", after=" << to_string(after);
|
|
|
|
// If we were proposing but aren't any longer, we need to reset the
|
|
// censorship tracking to avoid bogus warnings.
|
|
if ((before == ConsensusMode::Proposing || before == ConsensusMode::Observing) &&
|
|
before != after)
|
|
censorshipDetector_.reset();
|
|
|
|
mode_ = after;
|
|
|
|
// consensus.round is created before the engine applies the mode, so this is
|
|
// the first point where the round's mode is known. Every mode transition,
|
|
// including the one at round start, reaches here.
|
|
if (roundSpan_ && *roundSpan_)
|
|
{
|
|
roundSpan_->setAttribute(cs::attr::mode, toDisplayString(after).c_str());
|
|
}
|
|
}
|
|
|
|
json::Value
|
|
RCLConsensus::getJson(bool full) const
|
|
{
|
|
json::Value ret;
|
|
{
|
|
std::scoped_lock const _{mutex_};
|
|
ret = consensus_.getJson(full);
|
|
}
|
|
ret["validating"] = adaptor_.validating();
|
|
return ret;
|
|
}
|
|
|
|
void
|
|
RCLConsensus::timerEntry(
|
|
NetClock::time_point const& now,
|
|
std::unique_ptr<std::stringstream> const& clog)
|
|
{
|
|
try
|
|
{
|
|
std::scoped_lock const _{mutex_};
|
|
consensus_.timerEntry(now, clog);
|
|
}
|
|
catch (SHAMapMissingNode const& mn)
|
|
{
|
|
// This should never happen
|
|
std::stringstream ss;
|
|
ss << "During consensus timerEntry: " << mn.what();
|
|
JLOG(j_.error()) << ss.str();
|
|
CLOG(clog) << ss.str();
|
|
rethrow();
|
|
}
|
|
}
|
|
|
|
void
|
|
RCLConsensus::gotTxSet(NetClock::time_point const& now, RCLTxSet const& txSet)
|
|
{
|
|
try
|
|
{
|
|
std::scoped_lock const _{mutex_};
|
|
consensus_.gotTxSet(now, txSet);
|
|
}
|
|
catch (SHAMapMissingNode const& mn)
|
|
{
|
|
// This should never happen
|
|
JLOG(j_.error()) << "During consensus gotTxSet: " << mn.what();
|
|
rethrow();
|
|
}
|
|
}
|
|
|
|
/**
|
|
* @see Consensus::simulate
|
|
*/
|
|
|
|
void
|
|
RCLConsensus::simulate(
|
|
NetClock::time_point const& now,
|
|
std::optional<std::chrono::milliseconds> consensusDelay)
|
|
{
|
|
std::scoped_lock const _{mutex_};
|
|
consensus_.simulate(now, consensusDelay);
|
|
}
|
|
|
|
bool
|
|
RCLConsensus::peerProposal(NetClock::time_point const& now, RCLCxPeerPos const& newProposal)
|
|
{
|
|
std::scoped_lock const _{mutex_};
|
|
return consensus_.peerProposal(now, newProposal);
|
|
}
|
|
|
|
bool
|
|
RCLConsensus::Adaptor::preStartRound(RCLCxLedger const& prevLgr, hash_set<NodeID> const& nowTrusted)
|
|
{
|
|
// We have a key, we do not want out of sync validations after a restart
|
|
// and are not amendment blocked.
|
|
validating_ = validatorKeys_.keys && prevLgr.seq() >= app_.getMaxDisallowedLedger() &&
|
|
!app_.getOPs().isBlocked();
|
|
|
|
// If we are not running in standalone mode and there's a configured UNL,
|
|
// check to make sure that it's not expired.
|
|
if (validating_ && !app_.config().standalone() && (app_.getValidators().count() != 0u))
|
|
{
|
|
auto const when = app_.getValidators().expires();
|
|
|
|
if (!when || *when < app_.getTimeKeeper().now())
|
|
{
|
|
JLOG(j_.error()) << "Voluntarily bowing out of consensus process "
|
|
"because of an expired validator list.";
|
|
validating_ = false;
|
|
}
|
|
}
|
|
|
|
bool const synced = app_.getOPs().getOperatingMode() == OperatingMode::FULL;
|
|
|
|
if (validating_)
|
|
{
|
|
JLOG(j_.info()) << "Entering consensus process, validating, synced="
|
|
<< (synced ? "yes" : "no");
|
|
}
|
|
else
|
|
{
|
|
// Otherwise we just want to monitor the validation process.
|
|
JLOG(j_.info()) << "Entering consensus process, watching, synced="
|
|
<< (synced ? "yes" : "no");
|
|
}
|
|
|
|
// Notify inbound ledgers that we are starting a new round
|
|
inboundTransactions_.newRound(prevLgr.seq());
|
|
|
|
// Notify NegativeUNLVote that new validators are added
|
|
if (!nowTrusted.empty())
|
|
nUnlVote_.newValidators(prevLgr.seq() + 1, nowTrusted);
|
|
|
|
startRoundTracing(prevLgr);
|
|
|
|
// propose only if we're in sync with the network (and validating)
|
|
return validating_ && synced;
|
|
}
|
|
|
|
bool
|
|
RCLConsensus::Adaptor::haveValidated() const
|
|
{
|
|
return ledgerMaster_.haveValidated();
|
|
}
|
|
|
|
LedgerIndex
|
|
RCLConsensus::Adaptor::getValidLedgerIndex() const
|
|
{
|
|
return ledgerMaster_.getValidLedgerIndex();
|
|
}
|
|
|
|
std::pair<std::size_t, hash_set<RCLConsensus::Adaptor::NodeKey_t>>
|
|
RCLConsensus::Adaptor::getQuorumKeys() const
|
|
{
|
|
return app_.getValidators().getQuorumKeys();
|
|
}
|
|
|
|
std::size_t
|
|
RCLConsensus::Adaptor::laggards(
|
|
Ledger_t::Seq const seq,
|
|
hash_set<RCLConsensus::Adaptor::NodeKey_t>& trustedKeys) const
|
|
{
|
|
return app_.getValidations().laggards(seq, trustedKeys);
|
|
}
|
|
|
|
bool
|
|
RCLConsensus::Adaptor::validator() const
|
|
{
|
|
return validatorKeys_.keys.has_value();
|
|
}
|
|
|
|
void
|
|
RCLConsensus::Adaptor::updateOperatingMode(std::size_t const positions) const
|
|
{
|
|
if ((positions == 0u) && app_.getOPs().isFull())
|
|
app_.getOPs().setMode(OperatingMode::CONNECTED);
|
|
}
|
|
|
|
// Neither is static: both guarded bodies read the span-context members. With
|
|
// telemetry compiled out one body is empty and the other returns std::nullopt,
|
|
// so clang-tidy sees two methods that could be static.
|
|
// NOLINTBEGIN(readability-convert-member-functions-to-static)
|
|
void
|
|
RCLConsensus::Adaptor::startRoundTracing(RCLCxLedger const& prevLgr)
|
|
{
|
|
// The whole body is telemetry: every member it touches exists only to carry
|
|
// span state. It is compiled out rather than left to the early return below,
|
|
// because the work above that return — two virtual Telemetry calls and the
|
|
// strategy string compare — would otherwise run once per round for nothing.
|
|
#ifdef XRPL_ENABLE_TELEMETRY
|
|
namespace cs = telemetry::consensus::span;
|
|
|
|
// Capture the prior round's context BEFORE the new span overwrites
|
|
// roundSpanContext_; used to add a follows-from link from the new
|
|
// round's span so consecutive rounds remain navigable.
|
|
prevRoundSpanContext_ = roundSpanContext_;
|
|
|
|
// Reset the prior accept context so a stale value can't be used to
|
|
// link a validation that fires before this round's accept span exists.
|
|
acceptSpanContext_ = telemetry::SpanContext{};
|
|
|
|
if (roundSpan_)
|
|
roundSpan_.reset();
|
|
|
|
auto const strategy = app_.getTelemetry().getConsensusTraceStrategy();
|
|
|
|
telemetry::SpanContext const* const link =
|
|
prevRoundSpanContext_.isValid() ? &prevRoundSpanContext_ : nullptr;
|
|
|
|
if (strategy == telemetry::ConsensusTraceStrategy::Random)
|
|
{
|
|
// Experimental strategy, not used on a live network: each node gets a
|
|
// random trace_id, so one round arrives as one trace per node, joinable
|
|
// only by the consensus_ledger_id attribute. Still attach a follows-from
|
|
// link to the prior round so consecutive rounds stay navigable.
|
|
// linkedSpan is not TraceCategory-aware, so gate it explicitly to match
|
|
// the gating of the hashSpan/span factories used below.
|
|
if (link != nullptr && app_.getTelemetry().shouldTraceConsensus())
|
|
{
|
|
roundSpan_.emplace(telemetry::SpanGuard::linkedSpan(cs::round, *link));
|
|
}
|
|
else
|
|
{
|
|
roundSpan_.emplace(
|
|
telemetry::SpanGuard::span(
|
|
telemetry::TraceCategory::Consensus, telemetry::seg::consensus, cs::op::round));
|
|
}
|
|
}
|
|
else
|
|
{
|
|
// Deterministic (the default): derive the trace_id from the previous
|
|
// ledger hash so all validators tracing the same round share one trace.
|
|
roundSpan_.emplace(
|
|
telemetry::SpanGuard::hashSpan(
|
|
telemetry::TraceCategory::Consensus,
|
|
cs::round,
|
|
prevLgr.id().data(),
|
|
prevLgr.id().kBytes,
|
|
link));
|
|
}
|
|
|
|
if (!*roundSpan_)
|
|
{
|
|
return;
|
|
}
|
|
|
|
roundSpan_->setAttribute(cs::attr::ledgerId, to_string(prevLgr.id()).c_str());
|
|
roundSpan_->setAttribute(cs::attr::ledgerSeq, static_cast<int64_t>(prevLgr.seq()) + 1);
|
|
roundSpan_->setAttribute(cs::attr::traceStrategy, telemetry::strategyName(strategy));
|
|
roundSpan_->setAttribute(cs::attr::roundId, static_cast<int64_t>(prevLgr.seq()) + 1);
|
|
roundSpan_->setAttribute(cs::attr::previousLedgerSeq, static_cast<int64_t>(prevLgr.seq()));
|
|
roundSpan_->setAttribute(cs::attr::previousProposers, static_cast<int64_t>(prevProposers_));
|
|
roundSpan_->setAttribute(
|
|
cs::attr::previousRoundTimeMs, static_cast<int64_t>(prevRoundTime_.load().count()));
|
|
roundSpan_->setAttribute(cs::attr::consensusPhase, cs::val::phaseOpen);
|
|
|
|
// consensus_mode is stamped by onModeChange, which the engine calls just
|
|
// after this with the mode it is applying. Setting it here would record the
|
|
// previous round's mode.
|
|
|
|
roundSpan_->addEvent(cs::event::phaseOpen);
|
|
|
|
// roundSpanContext_ is the durable handle that child spans on other
|
|
// threads parent to. roundSpan_ is a thread-free SpanGuard that is
|
|
// reset() on a different worker than it was emplaced on, so spanContext()
|
|
// captures its own span and no scope work is needed.
|
|
roundSpanContext_ = roundSpan_->spanContext();
|
|
#endif
|
|
}
|
|
|
|
std::optional<telemetry::SpanGuard>
|
|
RCLConsensus::Adaptor::createValidationSpan()
|
|
{
|
|
// The whole body is telemetry: it only builds a span from stored contexts.
|
|
// Compiled out, it yields std::nullopt, so validate() takes neither branch
|
|
// that reads the ledger hash into a string.
|
|
#ifdef XRPL_ENABLE_TELEMETRY
|
|
namespace cs = telemetry::consensus::span;
|
|
|
|
// Prefer linking to the accept span (matches the design diagram and
|
|
// the "validation follows acceptance" causal model). Fall back to the
|
|
// round span only if the accept context isn't yet captured (e.g.
|
|
// tracing started after onAccept, or makeAcceptSpan returned a null
|
|
// guard).
|
|
if (acceptSpanContext_.isValid())
|
|
{
|
|
return telemetry::SpanGuard::linkedSpan(cs::validationSend, acceptSpanContext_);
|
|
}
|
|
|
|
if (!roundSpanContext_.isValid())
|
|
{
|
|
return std::nullopt;
|
|
}
|
|
|
|
return telemetry::SpanGuard::linkedSpan(cs::validationSend, roundSpanContext_);
|
|
#else
|
|
return std::nullopt;
|
|
#endif
|
|
}
|
|
// NOLINTEND(readability-convert-member-functions-to-static)
|
|
|
|
void
|
|
RCLConsensus::Adaptor::onPhaseEvent(std::string_view eventName, std::string_view phaseLabel)
|
|
{
|
|
namespace cs = telemetry::consensus::span;
|
|
|
|
if (!roundSpan_ || !*roundSpan_)
|
|
{
|
|
return;
|
|
}
|
|
|
|
roundSpan_->addEvent(eventName);
|
|
if (!phaseLabel.empty())
|
|
{
|
|
roundSpan_->setAttribute(cs::attr::consensusPhase, phaseLabel);
|
|
}
|
|
}
|
|
|
|
void
|
|
RCLConsensus::Adaptor::onOutcomeEvent(std::string_view eventName)
|
|
{
|
|
if (!roundSpan_ || !*roundSpan_)
|
|
{
|
|
return;
|
|
}
|
|
roundSpan_->addEvent(eventName);
|
|
}
|
|
|
|
void
|
|
RCLConsensus::startRound(
|
|
NetClock::time_point const& now,
|
|
RCLCxLedger::ID const& prevLgrId,
|
|
RCLCxLedger const& prevLgr,
|
|
hash_set<NodeID> const& nowUntrusted,
|
|
hash_set<NodeID> const& nowTrusted,
|
|
std::unique_ptr<std::stringstream> const& clog)
|
|
{
|
|
std::scoped_lock const _{mutex_};
|
|
consensus_.startRound(
|
|
now, prevLgrId, prevLgr, nowUntrusted, adaptor_.preStartRound(prevLgr, nowTrusted), clog);
|
|
}
|
|
|
|
RclConsensusLogger::RclConsensusLogger(char const* label, bool const validating, beast::Journal j)
|
|
: j_(j)
|
|
{
|
|
if (!validating && !j.info())
|
|
return;
|
|
start_ = std::chrono::steady_clock::now();
|
|
ss_ = std::make_unique<std::stringstream>();
|
|
header_ = "ConsensusLogger ";
|
|
header_ += label;
|
|
header_ += ": ";
|
|
}
|
|
|
|
RclConsensusLogger::~RclConsensusLogger()
|
|
{
|
|
if (!ss_)
|
|
return;
|
|
auto const duration = std::chrono::duration_cast<std::chrono::milliseconds>(
|
|
std::chrono::steady_clock::now() - start_);
|
|
std::stringstream outSs;
|
|
outSs << header_ << "duration " << (duration.count() / 1000) << '.' << std::setw(3)
|
|
<< std::setfill('0') << (duration.count() % 1000) << "s. " << ss_->str();
|
|
j_.sink().writeAlways(beast::Severity::Info, outSs.str());
|
|
}
|
|
|
|
} // namespace xrpl
|