Merge branch 'pratik/otel-phase4-consensus-tracing' into pratik/otel-phase5-docs-deployment

.github/scripts/levelization/results/loops.txt

@@ -4,6 +4,9 @@ Loop: test.jtx test.toplevel
 Loop: test.jtx test.unit_test
   test.unit_test ~= test.jtx
 
+Loop: xrpl.telemetry xrpld.consensus
+  xrpld.consensus ~= xrpl.telemetry
+
 Loop: xrpl.telemetry xrpld.rpc
   xrpld.rpc > xrpl.telemetry
 

.github/scripts/levelization/results/ordering.txt

@@ -47,6 +47,7 @@ libxrpl.shamap > xrpl.nodestore
 libxrpl.shamap > xrpl.protocol
 libxrpl.shamap > xrpl.shamap
 libxrpl.telemetry > xrpl.basics
+libxrpl.telemetry > xrpl.config
 libxrpl.telemetry > xrpl.telemetry
 libxrpl.tx > xrpl.basics
 libxrpl.tx > xrpl.conditions
@@ -251,8 +252,7 @@ xrpl.server > xrpl.shamap
 xrpl.shamap > xrpl.basics
 xrpl.shamap > xrpl.nodestore
 xrpl.shamap > xrpl.protocol
-xrpl.telemetry > xrpl.basics
-xrpl.telemetry > xrpld.consensus
+xrpl.telemetry > xrpl.config
 xrpl.tx > xrpl.basics
 xrpl.tx > xrpl.core
 xrpl.tx > xrpl.ledger
@@ -279,7 +279,6 @@ xrpld.consensus > xrpl.basics
 xrpld.consensus > xrpl.json
 xrpld.consensus > xrpl.ledger
 xrpld.consensus > xrpl.protocol
-xrpld.consensus > xrpl.telemetry
 xrpld.core > xrpl.basics
 xrpld.core > xrpl.config
 xrpld.core > xrpl.core
@@ -331,4 +330,5 @@ xrpld.shamap > xrpld.core
 xrpld.shamap > xrpl.protocol
 xrpld.shamap > xrpl.shamap
 xrpld.telemetry > xrpl.basics
+xrpld.telemetry > xrpld.consensus
 xrpld.telemetry > xrpl.telemetry

cmake/XrplCore.cmake

@@ -206,7 +206,7 @@ target_link_libraries(
 add_module(xrpl telemetry)
 target_link_libraries(
     xrpl.libxrpl.telemetry
-    PUBLIC xrpl.libxrpl.basics xrpl.libxrpl.beast
+    PUBLIC xrpl.libxrpl.basics xrpl.libxrpl.beast xrpl.libxrpl.config
 )
 if(telemetry)
     target_link_libraries(

conan.lock

@@ -16,7 +16,7 @@
         "nlohmann_json/3.11.3#45828be26eb619a2e04ca517bb7b828d%1701220705.259",
         "lz4/1.10.0#59fc63cac7f10fbe8e05c7e62c2f3504%1765850143.914",
         "libiconv/1.17#1e65319e945f2d31941a9d28cc13c058%1765842973.492",
-        "libcurl/8.20.0#465ac276192c197ddc6a9f4494004278%1779353234.048",
+        "libcurl/8.20.0#c90b0c91a33d9a79b519c1c70bafc823%1780907438.587",
         "libbacktrace/cci.20210118#a7691bfccd8caaf66309df196790a5a1%1765842973.03",
         "libarchive/3.8.7#c446109bd1f1d8ba7936c94189bc50e6%1778091117.848",
         "jemalloc/5.3.1#1fc58d55316041f10fbc1e8a2eae632a%1776700028.228",

docker/telemetry/grafana/provisioning/datasources/tempo.yaml

@@ -97,7 +97,7 @@ datasources:
             tag: command
             operator: "="
             scope: span
-            type: static
+            type: dynamic
           - id: rpc-status
             tag: rpc_status
             operator: "="
@@ -126,17 +126,17 @@ datasources:
             type: dynamic
           # Phase 4: Consensus tracing filters
           - id: consensus-mode
-            tag: xrpl.consensus.mode
+            tag: consensus_mode
             operator: "="
             scope: span
             type: static
           - id: consensus-round
-            tag: xrpl.consensus.round
+            tag: consensus_round
             operator: "="
             scope: span
             type: dynamic
           - id: consensus-ledger-seq
-            tag: xrpl.ledger.seq
+            tag: ledger_seq
             operator: "="
             scope: span
             type: static

include/xrpl/telemetry/Telemetry.h

@@ -83,8 +83,8 @@
     The OTel SDK's TracerProvider and Tracer are internally thread-safe.
 */
 
-#include <xrpl/basics/BasicConfig.h>
 #include <xrpl/beast/utility/Journal.h>
+#include <xrpl/config/BasicConfig.h>
 
 #include <atomic>
 #include <chrono>
@@ -148,11 +148,11 @@ public:
         std::string serviceName = "xrpld";
 
         /** OTel resource attribute `service.version` (set from BuildInfo). */
-        std::string serviceVersion{};
+        std::string serviceVersion;
 
         /** OTel resource attribute `service.instance.id` (defaults to node
             public key). */
-        std::string serviceInstanceId{};
+        std::string serviceInstanceId;
 
         /** OTLP/HTTP endpoint URL where spans are sent. */
         std::string exporterEndpoint = "http://localhost:4318/v1/traces";
@@ -161,7 +161,7 @@ public:
         bool useTls = false;
 
         /** Path to a CA certificate bundle for TLS verification. */
-        std::string tlsCertPath{};
+        std::string tlsCertPath;
 
         /** Path to this node's client certificate (PEM), presented to the
             collector for mutual TLS. Empty disables client-side auth, in

src/libxrpl/telemetry/SpanGuard.cpp

@@ -504,6 +504,14 @@ SpanGuard::discard()
     {
         gTlDiscardCurrentSpan = true;
         impl_->span->End();
+        // Clear here so discard() owns the flag's whole lifetime
+        // (set -> End -> clear) in one scope, rather than relying on
+        // FilteringSpanProcessor::OnEnd() to clear it. Today every valid guard
+        // wraps a recording span (head sampling is 1.0), so OnEnd() always runs
+        // and clearing here is equivalent — but colocating set and clear keeps
+        // the flag leak-proof if a later phase can hand back a non-recording
+        // span (e.g. honoring a non-sampled remote parent during propagation).
+        gTlDiscardCurrentSpan = false;
         impl_->span = nullptr;  // prevent ~Impl from calling End() again
         impl_.reset();
     }

src/libxrpl/telemetry/Telemetry.cpp

@@ -123,8 +123,7 @@ public:
         {
             // SpanGuard::discard() set the flag on this thread just before
             // calling Span::End(), which invokes OnEnd() synchronously.
-            // Clear the flag and drop the span.
-            gTlDiscardCurrentSpan = false;
+            // Drop the span.
             return;
         }
         delegate_->OnEnd(std::move(span));

src/libxrpl/telemetry/TelemetryConfig.cpp

@@ -7,7 +7,7 @@
     See cfg/xrpld-example.cfg for the full list of available options.
 */
 
-#include <xrpl/basics/BasicConfig.h>
+#include <xrpl/config/BasicConfig.h>
 #include <xrpl/telemetry/Telemetry.h>
 
 #include <chrono>

src/tests/libxrpl/telemetry/SpanGuardFactory.cpp

@@ -30,8 +30,12 @@ TEST(SpanGuardFactory, category_span_returns_null_when_disabled)
     auto span = SpanGuard::span(TraceCategory::Rpc, "rpc", "test");
     EXPECT_FALSE(span);
 
-    span.setAttribute("xrpl.rpc.command", "test");
-    span.setAttribute("xrpl.rpc.status", "success");
+    // Attribute keys use the underscore convention for span attributes (the
+    // dotted xrpl.<domain>. form is reserved for resource attributes). The
+    // canonical constants live in the xrpld-level *SpanNames.h headers, which a
+    // libxrpl test cannot include, so the keys are written as literals here.
+    span.setAttribute("command", "test");
+    span.setAttribute("rpc_status", "success");
 }
 
 TEST(SpanGuardFactory, child_span_null_when_no_parent)
@@ -85,23 +89,26 @@ TEST(SpanGuardFactory, discard_safe_on_null)
 
 TEST(SpanGuardFactory, consensus_close_time_attributes)
 {
-    // Verify the consensus attribute pattern compiles and
-    // doesn't crash with null SpanGuard.
+    // Verify the consensus attribute pattern compiles and doesn't crash with a
+    // null SpanGuard. Attribute keys/values use the underscore convention; the
+    // canonical consensus::span constants are defined in the xrpld-level
+    // ConsensusSpanNames.h, which a libxrpl test cannot include, so the keys are
+    // written as literals here.
     {
         auto span = telemetry::SpanGuard::span(
             telemetry::TraceCategory::Consensus, telemetry::seg::consensus, "accept.apply");
-        span.setAttribute("xrpl.consensus.ledger.seq", static_cast<int64_t>(42));
-        span.setAttribute("xrpl.consensus.close_time", static_cast<int64_t>(780000000));
-        span.setAttribute("xrpl.consensus.close_time_correct", true);
-        span.setAttribute("xrpl.consensus.close_resolution_ms", static_cast<int64_t>(30000));
-        span.setAttribute("xrpl.consensus.state", std::string("finished"));
-        span.setAttribute("xrpl.consensus.proposing", true);
-        span.setAttribute("xrpl.consensus.round_time_ms", static_cast<int64_t>(3500));
+        span.setAttribute("ledger_seq", static_cast<int64_t>(42));
+        span.setAttribute("close_time", static_cast<int64_t>(780000000));
+        span.setAttribute("close_time_correct", true);
+        span.setAttribute("close_resolution_ms", static_cast<int64_t>(30000));
+        span.setAttribute("consensus_state", std::string("finished"));
+        span.setAttribute("proposing", true);
+        span.setAttribute("round_time_ms", static_cast<int64_t>(3500));
     }
     {
         auto span = telemetry::SpanGuard::span(
             telemetry::TraceCategory::Consensus, telemetry::seg::consensus, "accept.apply");
-        span.setAttribute("xrpl.consensus.close_time_correct", false);
-        span.setAttribute("xrpl.consensus.state", std::string("moved_on"));
+        span.setAttribute("close_time_correct", false);
+        span.setAttribute("consensus_state", std::string("moved_on"));
     }
 }

src/tests/libxrpl/telemetry/TelemetryConfig.cpp

@@ -1,5 +1,5 @@
-#include <xrpl/basics/BasicConfig.h>
 #include <xrpl/beast/utility/Journal.h>
+#include <xrpl/config/BasicConfig.h>
 #include <xrpl/telemetry/Telemetry.h>
 
 #include <gtest/gtest.h>

src/xrpld/app/consensus/RCLConsensus.cpp

@@ -588,7 +588,8 @@ RCLConsensus::Adaptor::doAccept(
         static_cast<int64_t>(
             std::chrono::duration_cast<std::chrono::milliseconds>(closeResolution).count()));
     doAcceptSpan.setAttribute(
-        cs::attr::consensusState, std::string(consensusFail ? "moved_on" : "finished"));
+        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()));
@@ -1284,7 +1285,7 @@ RCLConsensus::Adaptor::startRoundTracing(RCLCxLedger const& prevLgr)
     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, "open");
+    roundSpan_->setAttribute(cs::attr::consensusPhase, cs::val::phaseOpen);
 
     roundSpan_->addEvent(cs::event::phaseOpen);
 

src/xrpld/consensus/Consensus.h

@@ -744,7 +744,9 @@ Consensus<Adaptor>::startRoundInternal(
     // after the new round span is in place.
     if (reason == StartRoundReason::Recovered)
     {
-        adaptor_.onPhaseEvent(telemetry::consensus::span::event::phaseOpen, "open");
+        adaptor_.onPhaseEvent(
+            telemetry::consensus::span::event::phaseOpen,
+            telemetry::consensus::span::val::phaseOpen);
     }
     mode_.set(mode, adaptor_);
     now_ = now;
@@ -994,7 +996,9 @@ Consensus<Adaptor>::simulate(
     result_->proposers = prevProposers_ = currPeerPositions_.size();
     prevRoundTime_ = result_->roundTime.read();
     phase_ = ConsensusPhase::Accepted;
-    adaptor_.onPhaseEvent(telemetry::consensus::span::event::phaseAccepted, "accepted");
+    adaptor_.onPhaseEvent(
+        telemetry::consensus::span::event::phaseAccepted,
+        telemetry::consensus::span::val::phaseAccepted);
     adaptor_.onForceAccept(
         *result_, previousLedger_, closeResolution_, rawCloseTimes_, mode_.get(), getJson(true));
     // NOLINTEND(bugprone-unchecked-optional-access)
@@ -1474,7 +1478,9 @@ Consensus<Adaptor>::phaseEstablish(std::unique_ptr<std::stringstream> const& clo
         }
     }
     phase_ = ConsensusPhase::Accepted;
-    adaptor_.onPhaseEvent(telemetry::consensus::span::event::phaseAccepted, "accepted");
+    adaptor_.onPhaseEvent(
+        telemetry::consensus::span::event::phaseAccepted,
+        telemetry::consensus::span::val::phaseAccepted);
     JLOG(j_.debug()) << "transitioned to ConsensusPhase::Accepted";
     adaptor_.onAccept(
         *result_,
@@ -1508,7 +1514,9 @@ Consensus<Adaptor>::closeLedger(std::unique_ptr<std::stringstream> const& clog)
     }
     openSpan_.reset();
     phase_ = ConsensusPhase::Establish;
-    adaptor_.onPhaseEvent(telemetry::consensus::span::event::phaseEstablish, "establish");
+    adaptor_.onPhaseEvent(
+        telemetry::consensus::span::event::phaseEstablish,
+        telemetry::consensus::span::val::phaseEstablish);
     JLOG(j_.debug()) << "transitioned to ConsensusPhase::Establish";
     rawCloseTimes_.self = now_;
     peerUnchangedCounter_ = 0;
@@ -1638,7 +1646,9 @@ Consensus<Adaptor>::updateOurPositions(std::unique_ptr<std::stringstream> const&
                 span.addEvent(
                     consensus::span::event::disputeResolve,
                     {{consensus::span::attr::txId, to_string(txId)},
-                     {consensus::span::attr::disputeOurVote, dispute.getOurVote() ? "yes" : "no"},
+                     {consensus::span::attr::disputeOurVote,
+                      dispute.getOurVote() ? std::string_view{consensus::span::val::yes}
+                                           : std::string_view{consensus::span::val::no}},
                      {consensus::span::attr::disputeYays, yaysStr},
                      {consensus::span::attr::disputeNays, naysStr}});
             }
@@ -1831,6 +1841,38 @@ Consensus<Adaptor>::haveConsensus(std::unique_ptr<std::stringstream> const& clog
         j_,
         clog);
 
+    // Set span attributes before the early-return branches below so the
+    // consensus.check span carries diagnostic data even when consensus is
+    // not reached (the No / Expired paths return early).
+    span.setAttribute(consensus::span::attr::agreeCount, static_cast<int64_t>(agree));
+    span.setAttribute(consensus::span::attr::disagreeCount, static_cast<int64_t>(disagree));
+    span.setAttribute(
+        consensus::span::attr::convergePercent, static_cast<int64_t>(convergePercent_));
+    span.setAttribute(consensus::span::attr::haveCloseTimeConsensus, haveCloseTimeConsensus_);
+    span.setAttribute(
+        consensus::span::attr::thresholdPercent,
+        static_cast<int64_t>(adaptor_.parms().avCtConsensusPct));
+    span.setAttribute(
+        consensus::span::attr::proposersFinished, static_cast<int64_t>(currentFinished));
+    span.setAttribute(consensus::span::attr::consensusStalled, stalled);
+    span.setAttribute(
+        consensus::span::attr::establishCounter, static_cast<int64_t>(establishCounter_));
+
+    std::string_view stateStr = consensus::span::val::no;
+    if (result_->state == ConsensusState::Yes)
+    {
+        stateStr = consensus::span::val::yes;
+    }
+    else if (result_->state == ConsensusState::MovedOn)
+    {
+        stateStr = consensus::span::val::movedOn;
+    }
+    else if (result_->state == ConsensusState::Expired)
+    {
+        stateStr = consensus::span::val::expired;
+    }
+    span.setAttribute(consensus::span::attr::consensusResult, stateStr);
+
     if (result_->state == ConsensusState::No)
     {
         CLOG(clog) << "No consensus. ";
@@ -1872,35 +1914,6 @@ Consensus<Adaptor>::haveConsensus(std::unique_ptr<std::stringstream> const& clog
         CLOG(clog) << "Unable to reach consensus " << json::Compact{getJson(true)} << ". ";
     }
 
-    span.setAttribute(consensus::span::attr::agreeCount, static_cast<int64_t>(agree));
-    span.setAttribute(consensus::span::attr::disagreeCount, static_cast<int64_t>(disagree));
-    span.setAttribute(
-        consensus::span::attr::convergePercent, static_cast<int64_t>(convergePercent_));
-    span.setAttribute(consensus::span::attr::haveCloseTimeConsensus, haveCloseTimeConsensus_);
-    span.setAttribute(
-        consensus::span::attr::thresholdPercent,
-        static_cast<int64_t>(adaptor_.parms().avCtConsensusPct));
-    span.setAttribute(
-        consensus::span::attr::proposersFinished, static_cast<int64_t>(currentFinished));
-    span.setAttribute(consensus::span::attr::consensusStalled, stalled);
-    span.setAttribute(
-        consensus::span::attr::establishCounter, static_cast<int64_t>(establishCounter_));
-
-    char const* stateStr = "no";
-    if (result_->state == ConsensusState::Yes)
-    {
-        stateStr = "yes";
-    }
-    else if (result_->state == ConsensusState::MovedOn)
-    {
-        stateStr = "moved_on";
-    }
-    else if (result_->state == ConsensusState::Expired)
-    {
-        stateStr = "expired";
-    }
-    span.setAttribute(consensus::span::attr::consensusResult, stateStr);
-
     CLOG(clog) << "Consensus has been reached. ";
     // NOLINTEND(bugprone-unchecked-optional-access)
     return true;

src/xrpld/consensus/ConsensusSpanNames.h

@@ -238,6 +238,10 @@ inline constexpr auto expired = makeStr("expired");
 inline constexpr auto increased = makeStr("increased");
 inline constexpr auto decreased = makeStr("decreased");
 inline constexpr auto unchanged = makeStr("unchanged");
+// consensus_phase attribute values (the phase the round is entering).
+inline constexpr auto phaseOpen = makeStr("open");
+inline constexpr auto phaseEstablish = makeStr("establish");
+inline constexpr auto phaseAccepted = makeStr("accepted");
 }  // namespace val
 
 }  // namespace xrpl::telemetry::consensus::span

src/xrpld/rpc/ServerHandler.h

@@ -177,7 +177,11 @@ private:
     void
     processSession(std::shared_ptr<Session> const&, std::shared_ptr<JobQueue::Coro> coro);
 
-    void
+    /** Process an RPC request and write the reply to `output`.
+        @return false if the request resulted in an error response, true
+                otherwise. Lets the caller's enclosing span reflect the outcome.
+    */
+    bool
     processRequest(
         Port const& port,
         std::string const& request,

src/xrpld/rpc/detail/RPCHandler.cpp

@@ -192,8 +192,17 @@ callMethod(JsonContext& context, Method method, std::string const& name, Object&
             rpc_span::attr::rpcStatus,
             ret ? std::string_view{rpc_span::val::error}
                 : std::string_view{rpc_span::val::success});
-        if (!ret)
+        // Reflect the result in the OTel span status, not just the attribute,
+        // so non-exception RPC errors (rpcTOO_BUSY, rpcNO_PERMISSION, ...) are
+        // visible to {status.code=error} queries.
+        if (ret)
+        {
+            span.setError(rpc_span::val::error);
+        }
+        else
+        {
             span.setOk();
+        }
         return ret;
     }
     catch (std::exception& e)
@@ -238,6 +247,7 @@ doCommand(RPC::JsonContext& context, json::Value& result)
         // "unknown" name only when the request truly omits both fields.
         auto span = SpanGuard::span(TraceCategory::Rpc, rpc_span::prefix::command, cmdName);
         span.setAttribute(rpc_span::attr::command, cmdName.c_str());
+        span.setAttribute(rpc_span::attr::rpcStatus, rpc_span::val::error);
         span.setError(getErrorInfo(error).token.cStr());
 
         injectError(error, result);

src/xrpld/rpc/detail/RpcSpanNames.h

@@ -159,7 +159,7 @@ using telemetry::attr_val::error;
 using telemetry::attr_val::success;
 inline constexpr auto admin = makeStr("admin");
 inline constexpr auto user = makeStr("user");
-inline constexpr auto unknownCommand = makeStr("unknown_command");
+inline constexpr auto unknownCommand = makeStr("unknown");
 /// "invalid_json" — WS message parse failure or oversize.
 inline constexpr auto invalidJson = makeStr("invalid_json");
 }  // namespace val

src/xrpld/rpc/detail/ServerHandler.cpp

@@ -444,6 +444,8 @@ ServerHandler::processSession(
         session->close({boost::beast::websocket::policy_error, "threshold exceeded"});
         // FIX: This rpcError is not delivered since the session
         // was just closed.
+        span.setAttribute(rpc_span::attr::rpcStatus, rpc_span::val::error);
+        span.setError("resource threshold exceeded");
         return rpcError(RpcSlowDown);
     }
 
@@ -475,6 +477,8 @@ ServerHandler::processSession(
                 jr[jss::api_version] = jv[jss::api_version];
 
             is->getConsumer().charge(Resource::kFeeMalformedRpc);
+            span.setAttribute(rpc_span::attr::rpcStatus, rpc_span::val::error);
+            span.setError(jr[jss::error].asString());
             return jr;
         }
 
@@ -555,12 +559,19 @@ ServerHandler::processSession(
         }
 
         jr[jss::request] = rq;
+        // Mark the span according to the final result. Doing it here (rather
+        // than an unconditional setOk later) ensures error responses — from
+        // doCommand, a FORBID role, or the catch block above — are not
+        // overwritten as OK.
+        span.setAttribute(rpc_span::attr::rpcStatus, rpc_span::val::error);
+        span.setError(jr[jss::error].asString());
     }
     else
     {
         if (jr[jss::result].isMember("forwarded") && jr[jss::result]["forwarded"])
             jr = jr[jss::result];
         jr[jss::status] = jss::success;
+        span.setOk();
     }
 
     if (jv.isMember(jss::id))
@@ -573,7 +584,6 @@ ServerHandler::processSession(
         jr[jss::api_version] = jv[jss::api_version];
 
     jr[jss::type] = jss::response;
-    span.setOk();
     return jr;
 }
 
@@ -589,7 +599,7 @@ ServerHandler::processSession(
     auto const requestBody = ::xrpl::buffersToString(session->request().body().data());
     span.setAttribute(rpc_span::attr::requestPayloadSize, static_cast<int64_t>(requestBody.size()));
 
-    processRequest(
+    bool const ok = processRequest(
         session->port(),
         requestBody,
         session->remoteAddress().atPort(0),
@@ -611,7 +621,15 @@ ServerHandler::processSession(
     {
         session->close(true);
     }
-    span.setOk();
+    // Reflect the request outcome on the wrapper span instead of always OK.
+    if (ok)
+    {
+        span.setOk();
+    }
+    else
+    {
+        span.setError(rpc_span::val::error);
+    }
 }
 
 static json::Value
@@ -630,7 +648,7 @@ constexpr json::Int kServerOverloaded = -32604;
 constexpr json::Int kForbidden = -32605;
 constexpr json::Int kWrongVersion = -32606;
 
-void
+bool
 ServerHandler::processRequest(
     Port const& port,
     std::string const& request,
@@ -643,18 +661,30 @@ ServerHandler::processRequest(
     auto span = SpanGuard::span(TraceCategory::Rpc, rpc_span::prefix::rpc, rpc_span::op::process);
     auto rpcJ = app_.getJournal("RPC");
 
+    // Tracks whether any failure occurred. Set on every error path (early
+    // returns, the catch block, and per-request error replies) and used at the
+    // end to mark the span status. The HTTP status code alone is insufficient:
+    // it stays 200 for batch responses and for ripplerpc < 3.0, so relying on
+    // it would let payload-level errors end the span as successful.
+    bool spanHadError = false;
+
+    // Marks the span as failed before sending an error reply, so the
+    // early-return validation paths below are not later seen as successful
+    // (the span would otherwise end UNSET, invisible to {status.code=error}).
+    auto httpReplyError = [&](int status, std::string const& message) {
+        spanHadError = true;
+        span.setError(message);
+        httpReply(status, message, output, rpcJ);
+    };
+
     json::Value jsonOrig;
     {
         json::Reader reader;
         if ((request.size() > RPC::Tuning::kMaxRequestSize) || !reader.parse(request, jsonOrig) ||
             !jsonOrig || !jsonOrig.isObject())
         {
-            httpReply(
-                400,
-                "Unable to parse request: " + reader.getFormattedErrorMessages(),
-                output,
-                rpcJ);
-            return;
+            httpReplyError(400, "Unable to parse request: " + reader.getFormattedErrorMessages());
+            return false;
         }
     }
 
@@ -665,8 +695,8 @@ ServerHandler::processRequest(
         batch = true;
         if (!jsonOrig.isMember(jss::params) || !jsonOrig[jss::params].isArray())
         {
-            httpReply(400, "Malformed batch request", output, rpcJ);
-            return;
+            httpReplyError(400, "Malformed batch request");
+            return false;
         }
         size = jsonOrig[jss::params].size();
     }
@@ -707,8 +737,8 @@ ServerHandler::processRequest(
         {
             if (!batch)
             {
-                httpReply(400, jss::invalid_API_version.cStr(), output, rpcJ);
-                return;
+                httpReplyError(400, jss::invalid_API_version.cStr());
+                return false;
             }
             json::Value r(json::ValueType::Object);
             r[jss::request] = jsonRPC;
@@ -750,8 +780,8 @@ ServerHandler::processRequest(
             {
                 if (!batch)
                 {
-                    httpReply(503, "Server is overloaded", output, rpcJ);
-                    return;
+                    httpReplyError(503, "Server is overloaded");
+                    return false;
                 }
                 json::Value r = jsonRPC;
                 r[jss::error] = makeJsonError(kServerOverloaded, "Server is overloaded");
@@ -765,8 +795,8 @@ ServerHandler::processRequest(
             usage.charge(Resource::kFeeMalformedRpc);
             if (!batch)
             {
-                httpReply(403, "Forbidden", output, rpcJ);
-                return;
+                httpReplyError(403, "Forbidden");
+                return false;
             }
             json::Value r = jsonRPC;
             r[jss::error] = makeJsonError(kForbidden, "Forbidden");
@@ -779,8 +809,8 @@ ServerHandler::processRequest(
             usage.charge(Resource::kFeeMalformedRpc);
             if (!batch)
             {
-                httpReply(400, "Null method", output, rpcJ);
-                return;
+                httpReplyError(400, "Null method");
+                return false;
             }
             json::Value r = jsonRPC;
             r[jss::error] = makeJsonError(kMethodNotFound, "Null method");
@@ -794,8 +824,8 @@ ServerHandler::processRequest(
             usage.charge(Resource::kFeeMalformedRpc);
             if (!batch)
             {
-                httpReply(400, "method is not string", output, rpcJ);
-                return;
+                httpReplyError(400, "method is not string");
+                return false;
             }
             json::Value r = jsonRPC;
             r[jss::error] = makeJsonError(kMethodNotFound, "method is not string");
@@ -809,8 +839,8 @@ ServerHandler::processRequest(
             usage.charge(Resource::kFeeMalformedRpc);
             if (!batch)
             {
-                httpReply(400, "method is empty", output, rpcJ);
-                return;
+                httpReplyError(400, "method is empty");
+                return false;
             }
             json::Value r = jsonRPC;
             r[jss::error] = makeJsonError(kMethodNotFound, "method is empty");
@@ -835,8 +865,8 @@ ServerHandler::processRequest(
             else if (!params.isArray() || params.size() != 1)
             {
                 usage.charge(Resource::kFeeMalformedRpc);
-                httpReply(400, "params unparsable", output, rpcJ);
-                return;
+                httpReplyError(400, "params unparsable");
+                return false;
             }
             else
             {
@@ -844,8 +874,8 @@ ServerHandler::processRequest(
                 if (!params.isObjectOrNull())
                 {
                     usage.charge(Resource::kFeeMalformedRpc);
-                    httpReply(400, "params unparsable", output, rpcJ);
-                    return;
+                    httpReplyError(400, "params unparsable");
+                    return false;
                 }
             }
         }
@@ -862,8 +892,8 @@ ServerHandler::processRequest(
                 usage.charge(Resource::kFeeMalformedRpc);
                 if (!batch)
                 {
-                    httpReply(400, "ripplerpc is not a string", output, rpcJ);
-                    return;
+                    httpReplyError(400, "ripplerpc is not a string");
+                    return false;
                 }
 
                 json::Value r = jsonRPC;
@@ -922,6 +952,7 @@ ServerHandler::processRequest(
                 << " when processing request: " << json::Compact{json::Value{params}};
             span.recordException(ex);
             span.setAttribute(rpc_span::attr::rpcStatus, rpc_span::val::error);
+            spanHadError = true;
             // LCOV_EXCL_STOP
         }
 
@@ -938,6 +969,7 @@ ServerHandler::processRequest(
         {
             if (result.isMember(jss::error))
             {
+                spanHadError = true;
                 result[jss::status] = jss::error;
                 result["code"] = result[jss::error_code];
                 result["message"] = result[jss::error_message];
@@ -958,6 +990,7 @@ ServerHandler::processRequest(
             // received.
             if (result.isMember(jss::error))
             {
+                spanHadError = true;
                 auto rq = params;
 
                 if (rq.isObject())
@@ -1053,8 +1086,20 @@ ServerHandler::processRequest(
         }
     }
 
-    span.setOk();
+    // Mark the span error if any request failed or the HTTP status is an error.
+    // spanHadError catches payload-level errors that httpStatus misses (batch
+    // responses and ripplerpc < 3.0 always return HTTP 200).
+    if (spanHadError || httpStatus >= 400)
+    {
+        span.setAttribute(rpc_span::attr::rpcStatus, rpc_span::val::error);
+        span.setError(rpc_span::val::error);
+    }
+    else
+    {
+        span.setOk();
+    }
     httpReply(httpStatus, response, output, rpcJ);
+    return !spanHadError;
 }
 
 //------------------------------------------------------------------------------

src/xrpld/telemetry/ConsensusReceiveTracing.h

@@ -31,25 +31,19 @@
  *      span.setAttribute(...);
  *  @endcode
  *
- *  @note These span names use inline string_view literals. When
- *  ConsensusSpanNames.h (from Phase 4) is available, callers should
- *  migrate to using the constexpr constants defined there.
+ *  @note Span names come from the canonical constants in
+ *  ConsensusSpanNames.h (consensus::span::proposalReceive /
+ *  validationReceive) so they stay in sync with the rest of Phase 4.
  */
 
+#include <xrpld/consensus/ConsensusSpanNames.h>
+
 #include <xrpl/proto/xrpl.pb.h>
 #include <xrpl/telemetry/SpanGuard.h>
 #include <xrpl/telemetry/TraceContextValidation.h>
 
 namespace xrpl::telemetry {
 
-// Inline span name constants for consensus receive spans.
-// Phase 4 will provide these via ConsensusSpanNames.h; these are
-// temporary definitions for the propagation infrastructure.
-namespace detail {
-inline constexpr std::string_view proposalReceiveName = "consensus.proposal.receive";
-inline constexpr std::string_view validationReceiveName = "consensus.validation.receive";
-}  // namespace detail
-
 /** Create a "consensus.proposal.receive" span for an incoming proposal.
  *
  *  If the message carries a TraceContext with a valid span_id, the
@@ -75,7 +69,7 @@ proposalReceiveSpan([[maybe_unused]] protocol::TMProposeSet const& msg)
             // trace_id so the receiving span shares the same trace.
             return SpanGuard::hashSpan(
                 TraceCategory::Consensus,
-                detail::proposalReceiveName,
+                consensus::span::proposalReceive,
                 reinterpret_cast<std::uint8_t const*>(tc.trace_id().data()),
                 tc.trace_id().size(),
                 reinterpret_cast<std::uint8_t const*>(tc.span_id().data()),
@@ -86,7 +80,8 @@ proposalReceiveSpan([[maybe_unused]] protocol::TMProposeSet const& msg)
     }
 #endif
     // No propagated context — create a standalone span.
-    return SpanGuard::span(TraceCategory::Consensus, "consensus", "proposal.receive");
+    return SpanGuard::span(
+        TraceCategory::Consensus, seg::consensus, consensus::span::op::proposalReceive);
 }
 
 /** Create a "consensus.validation.receive" span for an incoming validation.
@@ -111,7 +106,7 @@ validationReceiveSpan([[maybe_unused]] protocol::TMValidation const& msg)
         {
             return SpanGuard::hashSpan(
                 TraceCategory::Consensus,
-                detail::validationReceiveName,
+                consensus::span::validationReceive,
                 reinterpret_cast<std::uint8_t const*>(tc.trace_id().data()),
                 tc.trace_id().size(),
                 reinterpret_cast<std::uint8_t const*>(tc.span_id().data()),
@@ -122,7 +117,8 @@ validationReceiveSpan([[maybe_unused]] protocol::TMValidation const& msg)
     }
 #endif
     // No propagated context — create a standalone span.
-    return SpanGuard::span(TraceCategory::Consensus, "consensus", "validation.receive");
+    return SpanGuard::span(
+        TraceCategory::Consensus, seg::consensus, consensus::span::op::validationReceive);
 }
 
 }  // namespace xrpl::telemetry