Files
rippled/src/xrpld/app/consensus/RCLValidations.cpp
Pratik Mankawde 41b818b55b feat(telemetry): join a ledger's spans into one trace, add round histogram (WP-B3)
A slow fresh-sync ledger produced spans scattered across threads with no
way to relate them. They now share a trace id derived from the ledger's own
hash, the one value every participating site already holds, so nothing new
is plumbed across threads. This is the pattern the transaction pipeline
already uses for its tx id.

Joined: ledger.validate, ledger.store, and a new
consensus.validation.accept recorded when a trusted validation arrives. In
Tempo, searching one ledger hash returns them together, so an operator can
tell whether the ledger was slow to arrive, slow to be accepted, or slow to
be stored. They are siblings rather than a chain because the accept gate is
entered from three different threads, so no fixed parent order exists.

consensus.validation.accept also records why an arriving validation did or
did not advance the gate, which makes "validations arrive but are all
rejected" visible for the first time.

consensus_round_duration_ms turns the existing round-time span attribute
into a histogram, so a fleet trend needs a metric query rather than raw
trace inspection. An explicit bucket view is required, not optional: the
SDK default tops out at ten seconds while consensus abandons a round at two
minutes, so slow rounds would all fall in one bucket and every quantile
would read exactly ten seconds. Cost is one record per round.

Record layer: the histogram is native and needs no collector change. The
two new bounded attributes are added as span-metric dimensions to both
collector configs. The ledger hash stays out of them, since a per-ledger
dimension mints a series per ledger; it is indexed in Tempo as the join key.

The ledger.acquire span is not joined yet, because that file was being
changed concurrently. It is registered as an optional member of the join
group so nothing fails, and switching it is a one-line follow-up.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-07-25 20:19:25 +01:00

264 lines
9.1 KiB
C++

#include <xrpld/app/consensus/RCLValidations.h>
#include <xrpld/app/ledger/InboundLedger.h>
#include <xrpld/app/ledger/InboundLedgers.h>
#include <xrpld/app/ledger/LedgerMaster.h>
#include <xrpld/app/main/Application.h>
#include <xrpld/app/misc/ValidatorList.h>
#include <xrpld/consensus/ConsensusSpanNames.h>
#include <xrpld/consensus/Validations.h>
#include <xrpld/core/TimeKeeper.h>
#include <xrpl/basics/Log.h>
#include <xrpl/basics/chrono.h>
#include <xrpl/beast/utility/instrumentation.h>
#include <xrpl/core/Job.h>
#include <xrpl/core/JobQueue.h>
#include <xrpl/core/PerfLog.h>
#include <xrpl/protocol/Indexes.h>
#include <xrpl/protocol/PublicKey.h>
#include <xrpl/protocol/RippleLedgerHash.h>
#include <xrpl/protocol/SField.h>
#include <xrpl/protocol/tokens.h>
#include <xrpl/telemetry/SpanGuard.h>
#include <algorithm>
#include <memory>
#include <optional>
namespace xrpl {
RCLValidatedLedger::RCLValidatedLedger(MakeGenesis)
: ledgerID_{0}, ledgerSeq_{0}, j_{beast::Journal::getNullSink()}
{
}
RCLValidatedLedger::RCLValidatedLedger(
std::shared_ptr<Ledger const> const& ledger,
beast::Journal j)
: ledgerID_{ledger->header().hash}, ledgerSeq_{ledger->seq()}, j_{j}
{
auto const hashIndex = ledger->read(keylet::skip());
if (hashIndex)
{
XRPL_ASSERT(
hashIndex->getFieldU32(sfLastLedgerSequence) == (seq() - 1),
"xrpl::RCLValidatedLedger::RCLValidatedLedger(Ledger) : valid "
"last ledger sequence");
ancestors_ = hashIndex->getFieldV256(sfHashes).value();
}
else
{
JLOG(j_.warn()) << "Ledger " << ledgerSeq_ << ":" << ledgerID_
<< " missing recent ancestor hashes";
}
}
auto
RCLValidatedLedger::minSeq() const -> Seq
{
return seq() - std::min(seq(), static_cast<Seq>(ancestors_.size()));
}
auto
RCLValidatedLedger::seq() const -> Seq
{
return ledgerSeq_;
}
auto
RCLValidatedLedger::id() const -> ID
{
return ledgerID_;
}
auto
RCLValidatedLedger::operator[](Seq const& s) const -> ID
{
if (s >= minSeq() && s <= seq())
{
if (s == seq())
return ledgerID_;
Seq const diff = seq() - s;
return ancestors_[ancestors_.size() - diff];
}
JLOG(j_.warn()) << "Unable to determine hash of ancestor seq=" << s
<< " from ledger hash=" << ledgerID_ << " seq=" << ledgerSeq_
<< " (available: " << minSeq() << "-" << seq() << ")";
// Default ID that is less than all others
return ID{0};
}
// Return the sequence number of the earliest possible mismatching ancestor
RCLValidatedLedger::Seq
mismatch(RCLValidatedLedger const& a, RCLValidatedLedger const& b)
{
using Seq = RCLValidatedLedger::Seq;
// Find overlapping interval for known sequence for the ledgers
Seq const lower = std::max(a.minSeq(), b.minSeq());
Seq const upper = std::min(a.seq(), b.seq());
Seq curr = upper;
while (curr != Seq{0} && a[curr] != b[curr] && curr >= lower)
--curr;
// If the searchable interval mismatches entirely, then we have to
// assume the ledgers mismatch starting post genesis ledger
return (curr < lower) ? Seq{1} : (curr + Seq{1});
}
RCLValidationsAdaptor::RCLValidationsAdaptor(Application& app, beast::Journal j) : app_(app), j_(j)
{
}
NetClock::time_point
RCLValidationsAdaptor::now() const
{
return app_.getTimeKeeper().closeTime();
}
std::optional<RCLValidatedLedger>
RCLValidationsAdaptor::acquire(LedgerHash const& hash)
{
using namespace std::chrono_literals;
auto ledger = perf::measureDurationAndLog(
[&]() { return app_.getLedgerMaster().getLedgerByHash(hash); },
"getLedgerByHash",
10ms,
j_);
if (!ledger)
{
JLOG(j_.warn()) << "Need validated ledger for preferred ledger analysis " << hash;
Application* pApp = &app_;
app_.getJobQueue().addJob(JtAdvance, "GetConsL2", [pApp, hash, this]() {
JLOG(j_.debug()) << "JOB advanceLedger getConsensusLedger2 started";
pApp->getInboundLedgers().acquireAsync(hash, 0, InboundLedger::Reason::CONSENSUS);
});
return std::nullopt;
}
XRPL_ASSERT(
!ledger->open() && ledger->isImmutable(),
"xrpl::RCLValidationsAdaptor::acquire : valid ledger state");
XRPL_ASSERT(
ledger->header().hash == hash, "xrpl::RCLValidationsAdaptor::acquire : ledger hash match");
return RCLValidatedLedger(ledger, j_);
}
void
handleNewValidation(
Application& app,
std::shared_ptr<STValidation> const& val,
std::string const& source,
BypassAccept const bypassAccept,
std::optional<beast::Journal> j)
{
auto const& signingKey = val->getSignerPublic();
auto const& hash = val->getLedgerHash();
auto const seq = val->getFieldU32(sfLedgerSequence);
// Ensure validation is marked as trusted if signer currently trusted
auto masterKey = app.getValidators().getTrustedKey(signingKey);
if (!val->isTrusted() && masterKey)
val->setTrusted();
// If not currently trusted, see if signer is currently listed
if (!masterKey)
masterKey = app.getValidators().getListedKey(signingKey);
auto& validations = app.getValidations();
// masterKey is seated only if validator is trusted or listed
auto const outcome = validations.add(calcNodeID(masterKey.value_or(signingKey)), val);
if (outcome == ValStatus::Current)
{
if (val->isTrusted())
{
// Join this trusted validation to the trace of the ledger it
// validates. The acceptance decision it triggers runs in
// LedgerMaster::checkAccept, which may be this thread or another
// (checkAccept is also entered from the acquire completion job and
// from switchLCL), and it emits its own ledger.validate span. Both
// spans derive their trace id from the SAME validated-ledger hash
// via LedgerMaster::makeLedgerTraceSpan, so "which validation
// pushed this ledger over quorum, and how long did the resulting
// acceptance take" is one trace instead of two unrelated ones.
//
// Only TRUSTED validations get a span. Untrusted ones cannot move
// acceptance, so a span for them would be cost with no causality to
// show; this keeps the rate bounded by the UNL size per ledger, in
// line with the existing per-message consensus.validation.receive
// span rather than on top of it.
namespace cs = telemetry::consensus::span;
auto span = LedgerMaster::makeLedgerTraceSpan(cs::validationAccept, hash, seq);
span.setAttribute(
cs::attr::validationStatus, cs::validationStatusValue(static_cast<int>(outcome)));
span.setAttribute(cs::attr::fullValidation, val->isFull());
span.setAttribute(cs::attr::acceptGated, bypassAccept == BypassAccept::Yes);
// The span stays alive across checkAccept below, so its duration is
// the time this validation spent driving the acceptance decision --
// the number that grows when a node is slow to validate.
if (bypassAccept == BypassAccept::Yes)
{
XRPL_ASSERT(j, "xrpl::handleNewValidation : journal is available");
if (j.has_value())
{
JLOG(j->trace())
<< "Bypassing checkAccept for validation " << val->getLedgerHash();
}
}
else
{
app.getLedgerMaster().checkAccept(hash, seq);
}
}
return;
}
// Ensure that problematic validations from validators we trust are
// logged at the highest possible level.
//
// One might think that we should more than just log: we ought to also
// not relay validations that fail these checks. Alas, and somewhat
// counterintuitively, we *especially* want to forward such validations,
// so that our peers will also observe them and take independent notice of
// such validators, informing their operators.
if (auto const ls = val->isTrusted() ? validations.adaptor().journal().error()
: validations.adaptor().journal().info();
ls.active())
{
auto const id = [&masterKey, &signingKey]() {
auto ret = toBase58(TokenType::NodePublic, signingKey);
if (masterKey && masterKey != signingKey)
ret += ":" + toBase58(TokenType::NodePublic, *masterKey);
return ret;
}();
if (outcome == ValStatus::Conflicting)
{
ls << "Byzantine Behavior Detector: " << (val->isTrusted() ? "trusted " : "untrusted ")
<< id << ": Conflicting validation for " << seq << "!\n["
<< val->getSerializer().slice() << "]";
}
if (outcome == ValStatus::Multiple)
{
ls << "Byzantine Behavior Detector: " << (val->isTrusted() ? "trusted " : "untrusted ")
<< id << ": Multiple validations for " << seq << "/" << hash << "!\n["
<< val->getSerializer().slice() << "]";
}
}
}
} // namespace xrpl