merge: bring the review fixes forward from phase2-rpc-tracing

One conflict, in SpanGuard.h: this branch added struct TraceBytes and upstream
added enum SpanRole at the same position after TraceCategory. Unrelated
declarations, so both are kept.

src/libxrpl/telemetry/SpanGuard.cpp

@@ -86,7 +86,10 @@ SpanContext::SpanContext(std::shared_ptr<Impl> impl) : impl_(std::move(impl))
 bool
 SpanContext::isValid() const noexcept
 {
-    return impl_ != nullptr;
+    // Holding a Context is not proof of holding a span. GetCurrent() hands back
+    // an empty Context on a thread with no active span, and threadLocalContext()
+    // wraps that too. Ask the Context for its span instead of trusting impl_.
+    return impl_ != nullptr && otel_trace::GetSpan(impl_->ctx)->GetContext().IsValid();
 }
 
 // ===== SpanGuard::Impl ====================================================
@@ -174,11 +177,13 @@ namespace {
 constexpr char const* kLinkTypeKey = "link_type";
 constexpr char const* kLinkTypeFollowsFrom = "follows_from";
 
-// Map a TraceCategory to an OTel SpanKind so Tempo's service-graph /
-// RED metrics see the correct direction. RPC spans are emitted at the
-// server entry point (handler dispatch), Peer spans at inbound-message
-// receipt. Transactions / Consensus / Ledger are internal processing
-// and keep the default kInternal.
+// Per-category default OTel SpanKind, used when a call site passes no
+// SpanRole. A category cannot tell an inbound entry point from the
+// internal work under it, so RPC and Peer default to the entry-point
+// kind and any call site below the entry point passes SpanRole::Internal
+// instead. Transactions / Consensus / Ledger are internal throughout.
+// The kind drives direction in Tempo's service-graph / RED metrics,
+// which pair kServer with kClient and kConsumer with kProducer.
 otel_trace::SpanKind
 categoryToSpanKind(TraceCategory cat)
 {
@@ -196,6 +201,38 @@ categoryToSpanKind(TraceCategory cat)
     return otel_trace::SpanKind::kInternal;  // unreachable
 }
 
+/**
+ * Resolve the span kind to start a span with.
+ *
+ * An explicit SpanRole wins; SpanRole::FromCategory falls back to the
+ * category default above. Role and category are separate axes, so a single
+ * category can emit both an inbound handler and the internal work under it.
+ *
+ * @param cat  Trace subsystem category. Read only for SpanRole::FromCategory.
+ * @param role Role the caller asked for.
+ * @return The OTel span kind for this span.
+ */
+[[nodiscard]] otel_trace::SpanKind
+resolveSpanKind(TraceCategory cat, SpanRole role)
+{
+    switch (role)
+    {
+        case SpanRole::FromCategory:
+            return categoryToSpanKind(cat);
+        case SpanRole::Internal:
+            return otel_trace::SpanKind::kInternal;
+        case SpanRole::Server:
+            return otel_trace::SpanKind::kServer;
+        case SpanRole::Client:
+            return otel_trace::SpanKind::kClient;
+        case SpanRole::Producer:
+            return otel_trace::SpanKind::kProducer;
+        case SpanRole::Consumer:
+            return otel_trace::SpanKind::kConsumer;
+    }
+    return categoryToSpanKind(cat);  // unreachable
+}
+
 /**
  * Join a span-name prefix and suffix into the dotted full name.
  *
@@ -227,7 +264,11 @@ joinSpanName(std::string_view prefix, std::string_view name) noexcept
 }  // namespace
 
 SpanGuard
-SpanGuard::span(TraceCategory cat, std::string_view prefix, std::string_view name) noexcept
+SpanGuard::span(
+    TraceCategory cat,
+    std::string_view prefix,
+    std::string_view name,
+    SpanRole role) noexcept
 {
     auto* tel = Telemetry::getInstance();
     if ((tel == nullptr) || !tel->isEnabled() || !isCategoryEnabled(*tel, cat))
@@ -235,11 +276,15 @@ SpanGuard::span(TraceCategory cat, std::string_view prefix, std::string_view nam
     auto const fullName = joinSpanName(prefix, name);
     if (!fullName)
         return {};
-    return SpanGuard(std::make_unique<Impl>(tel->startSpan(*fullName, categoryToSpanKind(cat))));
+    return SpanGuard(std::make_unique<Impl>(tel->startSpan(*fullName, resolveSpanKind(cat, role))));
 }
 
 SpanGuard
-SpanGuard::freshRoot(TraceCategory cat, std::string_view prefix, std::string_view name) noexcept
+SpanGuard::freshRoot(
+    TraceCategory cat,
+    std::string_view prefix,
+    std::string_view name,
+    SpanRole role) noexcept
 {
     auto* tel = Telemetry::getInstance();
     if ((tel == nullptr) || !tel->isEnabled() || !isCategoryEnabled(*tel, cat))
@@ -250,7 +295,7 @@ SpanGuard::freshRoot(TraceCategory cat, std::string_view prefix, std::string_vie
     // Force a fresh trace root: do NOT inherit this thread's active span.
     auto rootCtx = opentelemetry::context::Context{otel_trace::kIsRootSpanKey, true};
     return SpanGuard(
-        std::make_unique<Impl>(tel->startSpan(*fullName, rootCtx, categoryToSpanKind(cat))));
+        std::make_unique<Impl>(tel->startSpan(*fullName, rootCtx, resolveSpanKind(cat, role))));
 }
 
 // ===== Child / linked span creation ========================================
@@ -630,8 +675,9 @@ ScopedSpanGuard::~ScopedSpanGuard()
 ScopedSpanGuard::ScopedSpanGuard(
     TraceCategory cat,
     std::string_view prefix,
-    std::string_view name) noexcept
-    : ScopedSpanGuard(SpanGuard::span(cat, prefix, name))
+    std::string_view name,
+    SpanRole role) noexcept
+    : ScopedSpanGuard(SpanGuard::span(cat, prefix, name, role))
 {
 }
 
@@ -639,9 +685,10 @@ ScopedSpanGuard
 ScopedSpanGuard::freshRoot(
     TraceCategory cat,
     std::string_view prefix,
-    std::string_view name) noexcept
+    std::string_view name,
+    SpanRole role) noexcept
 {
-    return ScopedSpanGuard(SpanGuard::freshRoot(cat, prefix, name));
+    return ScopedSpanGuard(SpanGuard::freshRoot(cat, prefix, name, role));
 }
 
 ScopedSpanGuard

src/libxrpl/telemetry/TelemetryConfig.cpp

@@ -8,11 +8,15 @@
  * See cfg/xrpld-example.cfg for the full list of available options.
  */
 
+#include <xrpl/basics/contract.h>
 #include <xrpl/config/BasicConfig.h>
 #include <xrpl/telemetry/Telemetry.h>
 
 #include <chrono>
 #include <cstdint>
+#include <limits>
+#include <optional>
+#include <stdexcept>
 #include <string>
 
 namespace xrpl::telemetry {
@@ -60,6 +64,71 @@ constexpr std::uint32_t batchDelayMs = 5000u;
 constexpr std::uint32_t maxQueueSize = 2048u;
 }  // namespace dflt
 
+/**
+ * Smallest accepted value for the three batch settings.
+ *
+ * All three size a queue or a timer, so zero is meaningless for every one of
+ * them. The OTel BatchSpanProcessor takes them as given and does not validate,
+ * so the config parser is the only place a nonsense value can be rejected.
+ */
+constexpr std::uint32_t kMinBatchSetting = 1u;
+
+/**
+ * Section name used in error messages, so the operator knows where to look.
+ */
+constexpr char const* kSectionLabel = "[telemetry]";
+
+/**
+ * Read a config value and reject anything outside minValue..UINT32_MAX.
+ *
+ * Section::get() lets boost::bad_lexical_cast escape. That derives from
+ * std::bad_cast, not std::runtime_error, so a mistyped value gives the operator
+ * a bare "bad cast" naming no key. Wrap it and rethrow with the key name.
+ *
+ * @param section The [telemetry] section to read from.
+ * @param name Key to read, as documented in cfg/xrpld-example.cfg.
+ * @param absentValue Value returned when the key is absent.
+ * @param minValue Smallest accepted value.
+ * @return The configured value, or absentValue if the key is absent.
+ * @note Throws std::runtime_error for a value that is not a whole number, and
+ * for one out of range, with a different message for each.
+ */
+[[nodiscard]] std::uint32_t
+readBounded(
+    Section const& section,
+    char const* name,
+    std::uint32_t absentValue,
+    std::uint32_t minValue)
+{
+    // Read as signed. boost::lexical_cast to an unsigned type wraps a leading
+    // minus instead of failing ("-1" yields 4294967295), so reading signed is
+    // the only way to see a negative value and reject it below.
+    std::optional<std::int64_t> parsed;
+    try
+    {
+        parsed = section.get<std::int64_t>(name);
+    }
+    catch (...)
+    {
+        Throw<std::runtime_error>(
+            std::string("Invalid value '") + name + "' in " + kSectionLabel +
+            ": must be a whole number.");
+    }
+
+    if (!parsed)
+        return absentValue;
+
+    constexpr auto maxValue = static_cast<std::int64_t>(std::numeric_limits<std::uint32_t>::max());
+    if (*parsed < static_cast<std::int64_t>(minValue) || *parsed > maxValue)
+    {
+        Throw<std::runtime_error>(
+            std::string("Invalid value '") + name + "' in " + kSectionLabel + ": must be between " +
+            std::to_string(minValue) + " and " + std::to_string(maxValue) + ".");
+    }
+
+    return static_cast<std::uint32_t>(*parsed);
+}
+
 /**
  * Derive a human-readable network type label from the numeric network ID.
  * @param networkId  The network identifier from [network_id] config.
@@ -108,10 +177,22 @@ makeTelemetrySetup(
     // traces; volume reduction is delegated to the collector's tail sampling.
     // setup.samplingRatio is a const member fixed at 1.0; nothing to parse.
 
-    setup.batchSize = section.valueOr<std::uint32_t>(key::batchSize, dflt::batchSize);
+    setup.batchSize = readBounded(section, key::batchSize, dflt::batchSize, kMinBatchSetting);
     setup.batchDelay = std::chrono::milliseconds{
-        section.valueOr<std::uint32_t>(key::batchDelayMs, dflt::batchDelayMs)};
-    setup.maxQueueSize = section.valueOr<std::uint32_t>(key::maxQueueSize, dflt::maxQueueSize);
+        readBounded(section, key::batchDelayMs, dflt::batchDelayMs, kMinBatchSetting)};
+    setup.maxQueueSize =
+        readBounded(section, key::maxQueueSize, dflt::maxQueueSize, kMinBatchSetting);
+
+    // The OTel SDK documents max_export_batch_size <= max_queue_size as a
+    // precondition of BatchSpanProcessorOptions and does not enforce it, so
+    // reject the pair here rather than hand the SDK a state it forbids.
+    if (setup.batchSize > setup.maxQueueSize)
+    {
+        Throw<std::runtime_error>(
+            std::string("Invalid value '") + key::batchSize + "' in " + kSectionLabel +
+            ": must not exceed '" + key::maxQueueSize + "' (" + std::to_string(setup.maxQueueSize) +
+            ").");
+    }
 
     setup.networkId = networkId;
     setup.networkType = networkTypeFromId(networkId);

src/tests/libxrpl/telemetry/TelemetryConfig.cpp

@@ -4,8 +4,85 @@
 
 #include <gtest/gtest.h>
 
+#include <chrono>
+#include <cstdint>
+#include <initializer_list>
+#include <limits>
+#include <stdexcept>
+#include <string>
+#include <utility>
+
 using namespace xrpl;
 
+namespace {
+
+/**
+ * Batch-setting keys of the [telemetry] section.
+ *
+ * Spelled once so every case below matches what the parser reads. A
+ * misspelling cannot hide: the accepting cases would see the default instead
+ * of the value they wrote, and the rejecting cases would stop rejecting.
+ */
+namespace key {
+constexpr char const* batchSize = "batch_size";
+constexpr char const* batchDelayMs = "batch_delay_ms";
+constexpr char const* maxQueueSize = "max_queue_size";
+}  // namespace key
+
+/**
+ * The upper bound quoted in the expected messages below.
+ *
+ * makeTelemetrySetup() derives it from std::uint32_t, so pin the literal to
+ * that type here rather than repeating an unanchored number in 5 messages.
+ */
+static_assert(std::numeric_limits<std::uint32_t>::max() == 4294967295u);
+
+using KeyValue = std::pair<char const*, char const*>;
+
+/**
+ * Parse a [telemetry] section holding only the given keys.
+ *
+ * A key that is not listed stays absent, so its default applies.
+ *
+ * @param values Key/value pairs to write into the section.
+ * @return The populated Setup struct.
+ */
+telemetry::Telemetry::Setup
+parseBatch(std::initializer_list<KeyValue> values)
+{
+    Section section;
+    for (auto const& [name, value] : values)
+        section.set(name, value);
+    return telemetry::makeTelemetrySetup(section, "nHUtest123", "2.0.0", 0);
+}
+
+/**
+ * Parse and return the rejection message.
+ *
+ * Only std::runtime_error is caught. A boost::bad_lexical_cast escaping the
+ * parser derives from std::bad_cast, so it propagates and fails the test
+ * instead of being mistaken for a clean rejection. That is the point of the
+ * not-a-number cases.
+ *
+ * @param values Key/value pairs to write into the section.
+ * @return The exception message, or "" if the parse succeeded.
+ */
+std::string
+batchRejection(std::initializer_list<KeyValue> values)
+{
+    try
+    {
+        static_cast<void>(parseBatch(values));
+        return {};
+    }
+    catch (std::runtime_error const& e)
+    {
+        return e.what();
+    }
+}
+
+}  // namespace
+
 TEST(TelemetryConfig, setup_defaults)
 {
     telemetry::Telemetry::Setup const s;
@@ -39,6 +116,11 @@ TEST(TelemetryConfig, parse_empty_section)
     EXPECT_EQ(setup.serviceVersion, "2.0.0");
     EXPECT_EQ(setup.serviceInstanceId, "nHUtest123");
     EXPECT_DOUBLE_EQ(setup.samplingRatio, 1.0);
+    // An absent key takes the documented default. setup_defaults covers the
+    // struct's own initializers; these three cover the parser applying them.
+    EXPECT_EQ(setup.batchSize, 512u);
+    EXPECT_EQ(setup.batchDelay, std::chrono::milliseconds{5000});
+    EXPECT_EQ(setup.maxQueueSize, 2048u);
     EXPECT_TRUE(setup.traceRpc);
     EXPECT_TRUE(setup.traceTransactions);
     EXPECT_TRUE(setup.traceConsensus);
@@ -83,6 +165,127 @@ TEST(TelemetryConfig, parse_full_section)
     EXPECT_FALSE(setup.traceLedger);
 }
 
+TEST(TelemetryConfig, batch_settings_accept_the_lower_bound_exactly)
+{
+    auto const setup =
+        parseBatch({{key::batchSize, "1"}, {key::batchDelayMs, "1"}, {key::maxQueueSize, "1"}});
+    EXPECT_EQ(setup.batchSize, 1u);
+    EXPECT_EQ(setup.batchDelay, std::chrono::milliseconds{1});
+    EXPECT_EQ(setup.maxQueueSize, 1u);
+}
+
+TEST(TelemetryConfig, batch_settings_accept_the_upper_bound_exactly)
+{
+    auto const setup = parseBatch(
+        {{key::batchSize, "4294967295"},
+         {key::batchDelayMs, "4294967295"},
+         {key::maxQueueSize, "4294967295"}});
+    EXPECT_EQ(setup.batchSize, 4294967295u);
+    EXPECT_EQ(setup.batchDelay, std::chrono::milliseconds{4294967295});
+    EXPECT_EQ(setup.maxQueueSize, 4294967295u);
+}
+
+TEST(TelemetryConfig, batch_size_zero_is_rejected)
+{
+    EXPECT_EQ(
+        batchRejection({{key::batchSize, "0"}}),
+        "Invalid value 'batch_size' in [telemetry]: must be between 1 and 4294967295.");
+}
+
+TEST(TelemetryConfig, batch_delay_ms_zero_is_rejected)
+{
+    EXPECT_EQ(
+        batchRejection({{key::batchDelayMs, "0"}}),
+        "Invalid value 'batch_delay_ms' in [telemetry]: must be between 1 and 4294967295.");
+}
+
+TEST(TelemetryConfig, max_queue_size_zero_is_rejected)
+{
+    EXPECT_EQ(
+        batchRejection({{key::maxQueueSize, "0"}}),
+        "Invalid value 'max_queue_size' in [telemetry]: must be between 1 and 4294967295.");
+}
+
+TEST(TelemetryConfig, batch_size_not_a_number_is_rejected_as_runtime_error)
+{
+    // Section::get() reaches boost::lexical_cast, which throws a std::bad_cast.
+    // Catching only std::runtime_error is the point: this fails unless the
+    // parser turned that into a message naming the key.
+    EXPECT_EQ(
+        batchRejection({{key::batchSize, "abc"}}),
+        "Invalid value 'batch_size' in [telemetry]: must be a whole number.");
+}
+
+TEST(TelemetryConfig, batch_delay_ms_not_a_number_is_rejected_as_runtime_error)
+{
+    EXPECT_EQ(
+        batchRejection({{key::batchDelayMs, "abc"}}),
+        "Invalid value 'batch_delay_ms' in [telemetry]: must be a whole number.");
+}
+
+TEST(TelemetryConfig, max_queue_size_not_a_number_is_rejected_as_runtime_error)
+{
+    EXPECT_EQ(
+        batchRejection({{key::maxQueueSize, "abc"}}),
+        "Invalid value 'max_queue_size' in [telemetry]: must be a whole number.");
+}
+
+TEST(TelemetryConfig, batch_size_fractional_is_rejected)
+{
+    // A batch counts spans, so "512.5" must not silently truncate to 512.
+    EXPECT_EQ(
+        batchRejection({{key::batchSize, "512.5"}}),
+        "Invalid value 'batch_size' in [telemetry]: must be a whole number.");
+}
+
+TEST(TelemetryConfig, batch_settings_reject_negative_rather_than_wrapping)
+{
+    // boost::lexical_cast to an unsigned type turns "-1" into 4294967295
+    // instead of failing, so a negative must land on the range check.
+    EXPECT_EQ(
+        batchRejection({{key::batchSize, "-1"}}),
+        "Invalid value 'batch_size' in [telemetry]: must be between 1 and 4294967295.");
+    EXPECT_EQ(
+        batchRejection({{key::batchDelayMs, "-1"}}),
+        "Invalid value 'batch_delay_ms' in [telemetry]: must be between 1 and 4294967295.");
+    EXPECT_EQ(
+        batchRejection({{key::maxQueueSize, "-1"}}),
+        "Invalid value 'max_queue_size' in [telemetry]: must be between 1 and 4294967295.");
+}
+
+TEST(TelemetryConfig, max_queue_size_above_the_upper_bound_is_rejected)
+{
+    EXPECT_EQ(
+        batchRejection({{key::maxQueueSize, "4294967296"}}),
+        "Invalid value 'max_queue_size' in [telemetry]: must be between 1 and 4294967295.");
+}
+
+TEST(TelemetryConfig, batch_size_above_max_queue_size_is_rejected)
+{
+    // The OTel SDK documents max_export_batch_size <= max_queue_size as a
+    // precondition and does not enforce it, so the parser must.
+    EXPECT_EQ(
+        batchRejection({{key::batchSize, "600"}, {key::maxQueueSize, "512"}}),
+        "Invalid value 'batch_size' in [telemetry]: must not exceed 'max_queue_size' (512).");
+}
+
+TEST(TelemetryConfig, batch_size_above_a_lowered_max_queue_size_is_rejected)
+{
+    // The likely operator mistake: lowering only max_queue_size and leaving
+    // batch_size at its 512 default.
+    EXPECT_EQ(
+        batchRejection({{key::maxQueueSize, "256"}}),
+        "Invalid value 'batch_size' in [telemetry]: must not exceed 'max_queue_size' (256).");
+}
+
+TEST(TelemetryConfig, batch_size_equal_to_max_queue_size_is_accepted)
+{
+    // The cross-check rejects only batchSize > maxQueueSize, so equal passes.
+    auto const setup = parseBatch({{key::batchSize, "512"}, {key::maxQueueSize, "512"}});
+    EXPECT_EQ(setup.batchSize, 512u);
+    EXPECT_EQ(setup.maxQueueSize, 512u);
+}
+
 TEST(TelemetryConfig, null_telemetry_factory)
 {
     telemetry::Telemetry::Setup setup;

src/xrpld/rpc/detail/RPCHandler.cpp

@@ -164,8 +164,10 @@ callMethod(JsonContext& context, Method method, std::string const& name, Object&
 {
     // Scoped so this command nests under rpc.process and becomes the ambient
     // parent of any command-internal spans (e.g. pathfind.request). Coro-aware
-    // storage keeps the scope correct across doRipplePathFind's yield.
-    auto span = ScopedSpanGuard(TraceCategory::Rpc, rpc_span::prefix::command, name);
+    // storage keeps the scope correct across doRipplePathFind's yield. Internal
+    // rather than Server: the inbound boundary is above rpc.process.
+    auto span =
+        ScopedSpanGuard(TraceCategory::Rpc, rpc_span::prefix::command, name, SpanRole::Internal);
     span.setAttribute(rpc_span::attr::command, name.c_str());
     span.setAttribute(rpc_span::attr::version, static_cast<int64_t>(context.apiVersion));
     span.setAttribute(
@@ -283,7 +285,9 @@ doCommand(rpc::JsonContext& context, json::Value& result)
         // registered handler names (plus "unknown") — see the helper for why
         // raw request input must not reach the telemetry pipeline.
         auto const cmdName = resolveCommandSpanName(context);
-        auto span = ScopedSpanGuard(TraceCategory::Rpc, rpc_span::prefix::command, cmdName);
+        // Internal for the same reason as the success path above.
+        auto span = ScopedSpanGuard(
+            TraceCategory::Rpc, rpc_span::prefix::command, cmdName, SpanRole::Internal);
         span.setAttribute(rpc_span::attr::command, cmdName);
         // Mirror the attribute set callMethod() puts on a successful command
         // span, so error spans stay filterable by API version and role.

src/xrpld/rpc/detail/ServerHandler.cpp

@@ -710,7 +710,9 @@ ServerHandler::processRequest(
     // yield in doRipplePathFind: the coro-aware context storage moves this
     // scope with the coroutine on resume (it is never stranded on a worker's
     // thread-local stack), so nesting and log-trace correlation both hold.
-    auto span = ScopedSpanGuard(TraceCategory::Rpc, rpc_span::prefix::rpc, rpc_span::op::process);
+    // Internal, not Server: the inbound boundary is rpc.http_request above.
+    auto span = ScopedSpanGuard(
+        TraceCategory::Rpc, rpc_span::prefix::rpc, rpc_span::op::process, SpanRole::Internal);
     auto rpcJ = app_.getJournal("RPC");
 
     // Tracks whether any failure occurred. Set on every error path (early