From 309d3e64495f262339bba70cfb7f82cbc6f4bd23 Mon Sep 17 00:00:00 2001 From: Pratik Mankawde <3397372+pratikmankawde@users.noreply.github.com> Date: Thu, 28 May 2026 20:15:41 +0100 Subject: [PATCH] fixed interround consensus linking and added some state attrs to spans. Signed-off-by: Pratik Mankawde <3397372+pratikmankawde@users.noreply.github.com> --- include/xrpl/telemetry/SpanGuard.h | 22 ++++-- src/libxrpl/telemetry/SpanGuard.cpp | 22 +++++- src/test/csf/Peer.h | 15 ++++ src/xrpld/app/consensus/RCLConsensus.cpp | 95 ++++++++++++++++++++++- src/xrpld/app/consensus/RCLConsensus.h | 54 +++++++++++++ src/xrpld/consensus/Consensus.h | 99 +++++++++++++++++++++--- src/xrpld/consensus/ConsensusSpanNames.h | 40 ++++++++++ src/xrpld/overlay/detail/PeerImp.cpp | 13 ++++ 8 files changed, 340 insertions(+), 20 deletions(-) diff --git a/include/xrpl/telemetry/SpanGuard.h b/include/xrpl/telemetry/SpanGuard.h index 60e32c4b8e..f394a23d7b 100644 --- a/include/xrpl/telemetry/SpanGuard.h +++ b/include/xrpl/telemetry/SpanGuard.h @@ -279,17 +279,22 @@ public: TraceCategory. All nodes using the same hash independently produce spans under the same trace_id, enabling cross-node correlation without context propagation. - @param cat Trace subsystem category. - @param name Full span name (e.g. "tx.receive"). - @param hashData Pointer to at least 16 bytes of hash data. - @param hashSize Size of the hash buffer (must be >= 16). + @param cat Trace subsystem category. + @param name Full span name (e.g. "tx.receive"). + @param hashData Pointer to at least 16 bytes of hash data. + @param hashSize Size of the hash buffer (must be >= 16). + @param followsFrom Optional captured context to attach as a + follows-from link. Use to stitch sequential + top-level spans (e.g. consecutive consensus + rounds). Ignored if nullptr or invalid. */ static SpanGuard hashSpan( TraceCategory const cat, std::string_view const name, std::uint8_t const* const hashData, - std::size_t const hashSize); + std::size_t const hashSize, + SpanContext const* followsFrom = nullptr); /** Create a hash-derived span with a remote parent. trace_id = hashData[0:16], parent span_id from protobuf context @@ -460,7 +465,12 @@ public: } [[nodiscard]] static SpanGuard - hashSpan(TraceCategory, std::string_view, std::uint8_t const*, std::size_t) + hashSpan( + TraceCategory, + std::string_view, + std::uint8_t const*, + std::size_t, + SpanContext const* = nullptr) { return {}; } diff --git a/src/libxrpl/telemetry/SpanGuard.cpp b/src/libxrpl/telemetry/SpanGuard.cpp index e09b9de123..b4d1dbd5c0 100644 --- a/src/libxrpl/telemetry/SpanGuard.cpp +++ b/src/libxrpl/telemetry/SpanGuard.cpp @@ -270,7 +270,8 @@ SpanGuard::hashSpan( TraceCategory const cat, std::string_view const name, std::uint8_t const* const hashData, - std::size_t const hashSize) + std::size_t const hashSize, + SpanContext const* const followsFrom) { if (hashSize < 16) return {}; @@ -293,6 +294,25 @@ SpanGuard::hashSpan( opentelemetry::nostd::shared_ptr( new otel_trace::DefaultSpan(syntheticCtx))); + if (followsFrom != nullptr && followsFrom->isValid()) + { + auto linkSpan = otel_trace::GetSpan(followsFrom->impl_->ctx); + if (linkSpan && linkSpan->GetContext().IsValid()) + { + auto tracer = tel->getTracer("xrpld"); + otel_trace::StartSpanOptions opts; + opts.parent = parentCtx; + opts.kind = categoryToSpanKind(cat); + return SpanGuard( + std::make_unique(tracer->StartSpan( + std::string(name), + {}, + {{linkSpan->GetContext(), + {{std::string(attr::linkType), std::string(attr_val::followsFrom)}}}}, + opts))); + } + } + return SpanGuard(std::make_unique(tel->startSpan(std::string(name), parentCtx))); } diff --git a/src/test/csf/Peer.h b/src/test/csf/Peer.h index f0b903e69e..6aa844389d 100644 --- a/src/test/csf/Peer.h +++ b/src/test/csf/Peer.h @@ -19,6 +19,7 @@ #include #include +#include namespace xrpl::test::csf { @@ -616,6 +617,20 @@ struct Peer { } + // Telemetry hooks — no-ops in the simulator. The generic engine calls + // these at every phase transition / outcome resolution so the + // production adaptor (RCLConsensus::Adaptor) can record events on the + // round span; the simulator runs without telemetry. + void + onPhaseEvent(std::string_view, std::string_view) + { + } + + void + onOutcomeEvent(std::string_view) + { + } + // Share a message by broadcasting to all connected peers template void diff --git a/src/xrpld/app/consensus/RCLConsensus.cpp b/src/xrpld/app/consensus/RCLConsensus.cpp index 525f437615..4a3bb6b4a8 100644 --- a/src/xrpld/app/consensus/RCLConsensus.cpp +++ b/src/xrpld/app/consensus/RCLConsensus.cpp @@ -349,6 +349,13 @@ RCLConsensus::Adaptor::onClose( telemetry::TraceCategory::Consensus, telemetry::seg::consensus, cs::op::ledgerClose); span.setAttribute(cs::attr::ledgerSeq, static_cast(ledger.ledger_->header().seq) + 1); span.setAttribute(cs::attr::mode, toDisplayString(mode).c_str()); + span.setAttribute( + cs::attr::txCountOpen, static_cast(app_.getOpenLedger().current()->txCount())); + span.setAttribute( + cs::attr::closeTimeResolutionMs, + static_cast( + std::chrono::duration_cast(ledger.closeTimeResolution()) + .count())); bool const wrongLCL = mode == ConsensusMode::wrongLedger; bool const proposing = mode == ConsensusMode::proposing; @@ -502,6 +509,15 @@ RCLConsensus::Adaptor::makeAcceptSpan(Result const& result) span->setAttribute( cs::attr::roundTimeMs, static_cast(result.roundTime.read().count())); span->setAttribute(cs::attr::quorum, static_cast(app_.getValidators().quorum())); + + // 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). + if (*span) + { + acceptSpanContext_ = span->captureContext(); + } return span; } @@ -567,6 +583,8 @@ RCLConsensus::Adaptor::doAccept( static_cast(rawCloseTimes.self.time_since_epoch().count())); doAcceptSpan.setAttribute( cs::attr::closeTimeVoteBins, static_cast(rawCloseTimes.peers.size())); + doAcceptSpan.setAttribute( + cs::attr::disputesResolvedCount, static_cast(result.disputes.size())); { auto const prevRes = prevLedger.closeTimeResolution(); auto const dir = [&]() -> std::string { @@ -908,9 +926,12 @@ RCLConsensus::Adaptor::validate(RCLCxLedger const& ledger, RCLTxSet const& txns, auto valSpan = createValidationSpan(); if (valSpan) { - valSpan->setAttribute( - telemetry::consensus::span::attr::ledgerSeq, static_cast(ledger.seq())); - valSpan->setAttribute(telemetry::consensus::span::attr::proposing, proposing); + namespace cs = telemetry::consensus::span; + valSpan->setAttribute(cs::attr::ledgerSeq, static_cast(ledger.seq())); + 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; @@ -920,6 +941,13 @@ RCLConsensus::Adaptor::validate(RCLCxLedger const& ledger, RCLTxSet const& txns, validationTime = lastValidationTime_ + 1s; lastValidationTime_ = validationTime; + if (valSpan) + { + valSpan->setAttribute( + telemetry::consensus::span::attr::validationSignTime, + static_cast(validationTime.time_since_epoch().count())); + } + if (!validatorKeys_.keys) { JLOG(j_.warn()) << "RCLConsensus::Adaptor::validate: ValidatorKeys " @@ -1191,11 +1219,23 @@ RCLConsensus::Adaptor::startRoundTracing(RCLCxLedger const& prevLgr) { 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 == "deterministic") { roundSpan_.emplace( @@ -1203,7 +1243,8 @@ RCLConsensus::Adaptor::startRoundTracing(RCLCxLedger const& prevLgr) telemetry::TraceCategory::Consensus, cs::round, prevLgr.id().data(), - prevLgr.id().bytes)); + prevLgr.id().bytes, + link)); } else { @@ -1220,6 +1261,13 @@ RCLConsensus::Adaptor::startRoundTracing(RCLCxLedger const& prevLgr) roundSpan_->setAttribute(cs::attr::mode, toDisplayString(mode_.load()).c_str()); roundSpan_->setAttribute(cs::attr::traceStrategy, strategy.c_str()); roundSpan_->setAttribute(cs::attr::roundId, static_cast(prevLgr.seq()) + 1); + roundSpan_->setAttribute(cs::attr::previousLedgerSeq, static_cast(prevLgr.seq())); + roundSpan_->setAttribute(cs::attr::previousProposers, static_cast(prevProposers_)); + roundSpan_->setAttribute( + cs::attr::previousRoundTimeMs, static_cast(prevRoundTime_.load().count())); + roundSpan_->setAttribute(cs::attr::consensusPhase, "open"); + + roundSpan_->addEvent(cs::event::phaseOpen); roundSpanContext_ = roundSpan_->captureContext(); } @@ -1229,12 +1277,51 @@ RCLConsensus::Adaptor::createValidationSpan() { 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_); } +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, diff --git a/src/xrpld/app/consensus/RCLConsensus.h b/src/xrpld/app/consensus/RCLConsensus.h index 63e440a24b..9b99f9df38 100644 --- a/src/xrpld/app/consensus/RCLConsensus.h +++ b/src/xrpld/app/consensus/RCLConsensus.h @@ -88,6 +88,34 @@ class RCLConsensus */ telemetry::SpanContext roundSpanContext_; + /** SpanContext of the prior round, captured before roundSpanContext_ + * is overwritten by the next round's startRoundTracing(). + * + * Used as a follows-from link target on the new round's span so + * consecutive rounds (each with its own deterministic trace-id) + * remain navigable in trace UIs as a connected sequence. + */ + telemetry::SpanContext prevRoundSpanContext_; + + /** SpanContext snapshot of the current round's accept span. + * + * Captured in makeAcceptSpan() and consumed by createValidationSpan() + * on the jtACCEPT worker thread so the validation.send span can be + * follows-from linked to consensus.accept (matching the design doc + * and span hierarchy diagram). Reset on each startRoundTracing() + * to prevent a stale prior-round context from being linked. + * + * Thread safety: same model as roundSpanContext_. The write in + * makeAcceptSpan happens on the main consensus thread under + * mutex_, and the read in createValidationSpan happens on the + * jtACCEPT worker thread that was posted by makeAcceptSpan. + * The job queue dispatch establishes a happens-before relation, + * so no atomic synchronization is needed. The next reset (in + * startRoundTracing) only runs after endConsensus → startRound, + * which by Consensus design only fires after doAccept completes. + */ + telemetry::SpanContext acceptSpanContext_; + public: using Ledger_t = RCLCxLedger; using NodeID_t = NodeID; @@ -197,6 +225,32 @@ class RCLConsensus std::optional createValidationSpan(); + /** Record a phase-transition event on the active round span. + * + * Called from the engine at each phase boundary + * (open/establish/accepted/recovery) so the round span carries a + * complete timeline of state changes. Also updates the + * `consensus_phase` attribute to the current phase name. + * + * @param eventName Event name (e.g. "phase.establish"). + * @param phaseLabel String value for the consensus_phase attr + * ("open"/"establish"/"accepted"). Empty to skip + * the attribute update (e.g. for "recovery"). + */ + void + onPhaseEvent(std::string_view eventName, std::string_view phaseLabel); + + /** Record a checkConsensus outcome event on the round span. + * + * Called from the engine at the establish→accepted transition so + * the path that drove acceptance (Yes / MovedOn / Expired) is + * queryable from the round-level trace. + * + * @param eventName Event name (e.g. "outcome.yes"). + */ + void + onOutcomeEvent(std::string_view eventName); + private: //--------------------------------------------------------------------- // The following members implement the generic Consensus requirements diff --git a/src/xrpld/consensus/Consensus.h b/src/xrpld/consensus/Consensus.h index 45d92714c9..e5f3b85d79 100644 --- a/src/xrpld/consensus/Consensus.h +++ b/src/xrpld/consensus/Consensus.h @@ -427,13 +427,28 @@ public: getJson(bool full) const; private: + /** Why startRoundInternal is being entered. + * + * Distinguishes the normal Initial entry (from public startRound) + * from a Recovered re-entry (from handleWrongLedger after the + * correct prior ledger was acquired mid-round). The Recovered path + * resets phase to Open within the SAME round, which is a state + * transition that would otherwise be invisible in traces — the + * recovery flag drives a `phase.recovery` event on the round span. + */ + enum class StartRoundReason : std::uint8_t { + Initial, + Recovered, + }; + void startRoundInternal( NetClock::time_point const& now, typename Ledger_t::ID const& prevLedgerID, Ledger_t const& prevLedger, ConsensusMode mode, - std::unique_ptr const& clog); + std::unique_ptr const& clog, + StartRoundReason reason = StartRoundReason::Initial); // Change our view of the previous ledger void @@ -693,8 +708,21 @@ Consensus::startRoundInternal( typename Ledger_t::ID const& prevLedgerID, Ledger_t const& prevLedger, ConsensusMode mode, - std::unique_ptr const& clog) + std::unique_ptr const& clog, + StartRoundReason const reason) { + // Recovery path: handleWrongLedger acquired the correct prior ledger + // and re-entered startRoundInternal mid-round. The roundSpan_ owned by + // the adaptor is still the SAME span as before — startRoundTracing is + // not called on the recovery path, so we record the recovery as an + // event on the surviving round span. Pass empty phaseLabel to leave + // consensus_phase unchanged (the actual phase reset to open is marked + // separately by the new openSpan_ being emplaced below). + if (reason == StartRoundReason::Recovered) + { + adaptor_.onPhaseEvent(telemetry::consensus::span::event::phaseRecovery, ""); + } + phase_ = ConsensusPhase::open; JLOG(j_.debug()) << "transitioned to ConsensusPhase::open "; CLOG(clog) << "startRoundInternal transitioned to ConsensusPhase::open, " @@ -709,6 +737,15 @@ Consensus::startRoundInternal( telemetry::TraceCategory::Consensus, telemetry::seg::consensus, telemetry::consensus::span::op::phaseOpen)); + // On the Recovered path, fire phase.open here because startRoundTracing + // (which fires it for the Initial path) is not called on re-entry. On + // the Initial path this is a no-op because the round span hasn't been + // created yet — the phase.open event is fired later by startRoundTracing + // after the new round span is in place. + if (reason == StartRoundReason::Recovered) + { + adaptor_.onPhaseEvent(telemetry::consensus::span::event::phaseOpen, "open"); + } mode_.set(mode, adaptor_); now_ = now; prevLedgerID_ = prevLedgerID; @@ -957,6 +994,7 @@ Consensus::simulate( result_->proposers = prevProposers_ = currPeerPositions_.size(); prevRoundTime_ = result_->roundTime.read(); phase_ = ConsensusPhase::accepted; + adaptor_.onPhaseEvent(telemetry::consensus::span::event::phaseAccepted, "accepted"); adaptor_.onForceAccept( *result_, previousLedger_, closeResolution_, rawCloseTimes_, mode_.get(), getJson(true)); // NOLINTEND(bugprone-unchecked-optional-access) @@ -1105,7 +1143,13 @@ Consensus::handleWrongLedger( { JLOG(j_.info()) << "Have the consensus ledger " << prevLedgerID_; CLOG(clog) << "Have the consensus ledger " << prevLedgerID_ << ". "; - startRoundInternal(now_, lgrId, *newLedger, ConsensusMode::switchedLedger, clog); + startRoundInternal( + now_, + lgrId, + *newLedger, + ConsensusMode::switchedLedger, + clog, + StartRoundReason::Recovered); } else { @@ -1414,7 +1458,23 @@ Consensus::phaseEstablish(std::unique_ptr const& clo prevProposers_ = currPeerPositions_.size(); prevRoundTime_ = result_->roundTime.read(); endEstablishTracing(); + { + namespace cs = telemetry::consensus::span; + if (result_->state == ConsensusState::Yes) + { + adaptor_.onOutcomeEvent(cs::event::outcomeYes); + } + else if (result_->state == ConsensusState::MovedOn) + { + adaptor_.onOutcomeEvent(cs::event::outcomeMovedOn); + } + else if (result_->state == ConsensusState::Expired) + { + adaptor_.onOutcomeEvent(cs::event::outcomeExpired); + } + } phase_ = ConsensusPhase::accepted; + adaptor_.onPhaseEvent(telemetry::consensus::span::event::phaseAccepted, "accepted"); JLOG(j_.debug()) << "transitioned to ConsensusPhase::accepted"; adaptor_.onAccept( *result_, @@ -1434,8 +1494,21 @@ Consensus::closeLedger(std::unique_ptr const& clog) // We should not be closing if we already have a position XRPL_ASSERT(!result_, "xrpl::Consensus::closeLedger : result is not set"); + // Annotate the open-phase span with end-of-phase metadata before + // ending it, so a Tempo/Jaeger query for consensus.phase.open shows + // how long the phase ran and how many peer positions arrived during it. + openTime_.tick(clock_.now()); + if (openSpan_ && *openSpan_) + { + namespace cs = telemetry::consensus::span; + openSpan_->setAttribute( + cs::attr::openDurationMs, static_cast(openTime_.read().count())); + openSpan_->setAttribute( + cs::attr::peerPositionsAtClose, static_cast(currPeerPositions_.size())); + } openSpan_.reset(); phase_ = ConsensusPhase::establish; + adaptor_.onPhaseEvent(telemetry::consensus::span::event::phaseEstablish, "establish"); JLOG(j_.debug()) << "transitioned to ConsensusPhase::establish"; rawCloseTimes_.self = now_; peerUnchangedCounter_ = 0; @@ -1807,6 +1880,11 @@ Consensus::haveConsensus(std::unique_ptr const& clog span.setAttribute( consensus::span::attr::thresholdPercent, static_cast(adaptor_.parms().avCT_CONSENSUS_PCT)); + span.setAttribute( + consensus::span::attr::proposersFinished, static_cast(currentFinished)); + span.setAttribute(consensus::span::attr::consensusStalled, stalled); + span.setAttribute( + consensus::span::attr::establishCounter, static_cast(establishCounter_)); char const* stateStr = "no"; if (result_->state == ConsensusState::Yes) @@ -1963,13 +2041,16 @@ Consensus::updateEstablishTracing() { if (!establishSpan_) return; + namespace cs = telemetry::consensus::span; + establishSpan_->setAttribute(cs::attr::convergePercent, static_cast(convergePercent_)); + establishSpan_->setAttribute(cs::attr::establishCount, static_cast(establishCounter_)); establishSpan_->setAttribute( - telemetry::consensus::span::attr::convergePercent, static_cast(convergePercent_)); - establishSpan_->setAttribute( - telemetry::consensus::span::attr::establishCount, static_cast(establishCounter_)); - establishSpan_->setAttribute( - telemetry::consensus::span::attr::proposers, - static_cast(currPeerPositions_.size())); + cs::attr::proposers, static_cast(currPeerPositions_.size())); + if (result_) + { + establishSpan_->setAttribute( + cs::attr::disputesCount, static_cast(result_->disputes.size())); + } } template diff --git a/src/xrpld/consensus/ConsensusSpanNames.h b/src/xrpld/consensus/ConsensusSpanNames.h index ed4c23d363..74a8a74cde 100644 --- a/src/xrpld/consensus/ConsensusSpanNames.h +++ b/src/xrpld/consensus/ConsensusSpanNames.h @@ -137,11 +137,37 @@ inline constexpr auto ledgerId = makeStr("consensus_ledger_id"); inline constexpr auto mode = makeStr("consensus_mode"); inline constexpr auto round = makeStr("consensus_round"); inline constexpr auto roundId = makeStr("consensus_round_id"); +/// Current phase name attached to consensus.round; updated on each +/// phase transition event (open/establish/accepted). +inline constexpr auto consensusPhase = makeStr("consensus_phase"); +/// Boolean flag set on consensus.check when checkConsensus reports stalled. +inline constexpr auto consensusStalled = makeStr("consensus_stalled"); /// Domain-owned bare attrs. inline constexpr auto proposers = makeStr("proposers"); inline constexpr auto roundTimeMs = makeStr("round_time_ms"); inline constexpr auto proposing = makeStr("proposing"); +/// Round continuity / context attrs (set on consensus.round at round start). +inline constexpr auto previousProposers = makeStr("previous_proposers"); +inline constexpr auto previousRoundTimeMs = makeStr("previous_round_time_ms"); +inline constexpr auto previousLedgerSeq = makeStr("previous_ledger_seq"); +inline constexpr auto closeTimeResolutionMs = makeStr("close_time_resolution_ms"); +/// Open-phase end metadata (set on consensus.phase.open before reset). +inline constexpr auto openDurationMs = makeStr("open_duration_ms"); +inline constexpr auto peerPositionsAtClose = makeStr("peer_positions_at_close"); +/// Ledger-close inputs. +inline constexpr auto txCountOpen = makeStr("tx_count_open"); +/// Establish/check additional state. +inline constexpr auto proposersFinished = makeStr("proposers_finished"); +inline constexpr auto establishCounter = makeStr("establish_counter"); +/// Accept/apply enrichment. +inline constexpr auto disputesResolvedCount = makeStr("disputes_resolved_count"); +/// Validation send/receive enrichment. +inline constexpr auto fullValidation = makeStr("full_validation"); +inline constexpr auto validationSignTime = makeStr("validation_sign_time"); +/// Receive-side hash prefixes for cross-peer correlation. +inline constexpr auto prevLedgerPrefix = makeStr("prev_ledger_prefix"); +inline constexpr auto positionHashPrefix = makeStr("position_hash_prefix"); /// "consensus_state" — domain-qualified (collides with other domains' state). inline constexpr auto consensusState = makeStr("consensus_state"); inline constexpr auto parentCloseTime = makeStr("parent_close_time"); @@ -180,6 +206,20 @@ namespace event { inline constexpr auto disputeResolve = join(makeStr("dispute"), makeStr("resolve")); /// "tx.included" inline constexpr auto txIncluded = join(makeStr("tx"), makeStr("included")); + +/// Phase transition events — fired on consensus.round at each transition +/// so the round-level span carries a complete timeline of phase changes, +/// including the handleWrongLedger recovery edge that re-enters Open. +inline constexpr auto phaseOpen = join(makeStr("phase"), makeStr("open")); +inline constexpr auto phaseEstablish = join(makeStr("phase"), makeStr("establish")); +inline constexpr auto phaseAccepted = join(makeStr("phase"), makeStr("accepted")); +inline constexpr auto phaseRecovery = join(makeStr("phase"), makeStr("recovery")); + +/// Outcome events — fired on consensus.round at the establish→accepted +/// transition so the path that drove acceptance is queryable. +inline constexpr auto outcomeYes = join(makeStr("outcome"), makeStr("yes")); +inline constexpr auto outcomeMovedOn = join(makeStr("outcome"), makeStr("moved_on")); +inline constexpr auto outcomeExpired = join(makeStr("outcome"), makeStr("expired")); } // namespace event // ===== Attribute values ====================================================== diff --git a/src/xrpld/overlay/detail/PeerImp.cpp b/src/xrpld/overlay/detail/PeerImp.cpp index 9faae9c83a..bfb3655bb9 100644 --- a/src/xrpld/overlay/detail/PeerImp.cpp +++ b/src/xrpld/overlay/detail/PeerImp.cpp @@ -1982,6 +1982,15 @@ PeerImp::onMessage(std::shared_ptr const& m) span->setAttribute(telemetry::consensus::span::attr::trusted, isTrusted); span->setAttribute( telemetry::consensus::span::attr::round, static_cast(set.proposeseq())); + // First 16 hex chars (8 bytes) of each hash — enough to disambiguate + // peer positions and prior ledgers without exporting full 32-byte + // hashes on every receive event. + span->setAttribute( + telemetry::consensus::span::attr::prevLedgerPrefix, + to_string(prevLedger).substr(0, 16).c_str()); + span->setAttribute( + telemetry::consensus::span::attr::positionHashPrefix, + to_string(proposeHash).substr(0, 16).c_str()); std::weak_ptr const weak = shared_from_this(); app_.getJobQueue().addJob( @@ -2572,6 +2581,10 @@ PeerImp::onMessage(std::shared_ptr const& m) telemetry::consensus::span::attr::ledgerSeq, static_cast(val->getFieldU32(sfLedgerSequence))); } + span->setAttribute(telemetry::consensus::span::attr::fullValidation, val->isFull()); + span->setAttribute( + telemetry::consensus::span::attr::validationSignTime, + static_cast(val->getSignTime().time_since_epoch().count())); if (!isTrusted && (tracking_.load() == Tracking::diverged)) {