feat(telemetry): trace each online-delete rotation phase and time it

Emits `nodestore.rotate` as a fresh trace root at the start of every rotation
in `SHAMapStoreImp::run`, with one child span per phase: `clear_prior`,
`copy`, `freshen.keys`, `freshen.fetch`, `new_backend`, `clear_caches`,
`swap`, and `health_wait`. `RotationPhase`'s destructor also records the
phase's wall-clock into `rotation_phase_duration_seconds{stage}`, so a
sampled trace and an exact histogram both reach Grafana.

The root carries `ledger_seq` and `last_rotated`; every phase carries a
count attribute the phase already computed (`node_count`, `key_count`,
`copy_forwards`) so nothing extra runs to satisfy telemetry. Outcome is one
of `complete|expired|stopping|missing_node`, stamped by a `RotationOutcome`
RAII helper on whichever exit runs first, and asserted in its destructor to
catch a new return path that forgot to name one.

`freshenCache` is split so `getKeys()` is timed apart from the fetch loop —
that split is what makes `freshen.keys` overlap the frozen receive handlers
in a Tempo view. The `health_wait` child opens only inside a running
rotation, so the pre-rotation `healthWait()` at the top of `run()` never
mints an orphan root.

src/xrpld/app/misc/SHAMapStoreImp.cpp

@@ -1,3 +1,4 @@
+// cspell:ignore ISTOGRAM Wreturn
 #include <xrpld/app/misc/SHAMapStoreImp.h>
 
 #include <xrpld/app/ledger/TransactionMaster.h>
@@ -403,6 +404,50 @@ SHAMapStoreImp::run()
         // will delete up to (not including) lastRotated
         if (readyToRotate)
         {
+            namespace ns = telemetry::nodestore_span;
+            namespace lv = telemetry::lval::rotation_phase;
+
+            // One trace per rotation. A fresh root: the SHAMapStore thread has
+            // no ambient span. Every phase below is a child through the
+            // thread's scope.
+            auto rotateSpan = telemetry::ScopedSpanGuard::freshRoot(
+                telemetry::TraceCategory::Ledger, telemetry::seg::nodestore, ns::op::rotate);
+            rotateSpan.setAttribute(ns::attr::ledgerSeq, static_cast<std::int64_t>(validatedSeq));
+            rotateSpan.setAttribute(ns::attr::lastRotated, static_cast<std::int64_t>(lastRotated));
+
+            // Stamp the outcome exactly once, on whichever exit runs first.
+            // The destructor asserts an exit was named: any exit without one
+            // is a new return path that forgot to. RAII so throw/continue paths
+            // still record the outcome.
+            struct RotationOutcome
+            {
+                telemetry::ScopedSpanGuard& span;
+                bool& rotating;
+                std::optional<ns::RotationExit> exit;
+                void
+                finish(ns::RotationExit e) noexcept
+                {
+                    if (exit)
+                        return;
+                    exit = e;
+                    span.setAttribute(ns::attr::outcome, ns::rotationOutcome(e));
+                }
+                ~RotationOutcome()
+                {
+                    XRPL_ASSERT(
+                        exit.has_value(),
+                        "xrpl::SHAMapStoreImp::run : rotation exit named an outcome");
+                    rotating = false;
+                }
+            };
+            RotationOutcome outcome{rotateSpan, rotating_, std::nullopt};
+            rotating_ = true;
+
+            auto const exitFor = [](HealthResult r) {
+                return r == HealthResult::Stopping ? ns::RotationExit::Stopping
+                                                   : ns::RotationExit::Expired;
+            };
+
             JLOG(journal_.warn()) << "rotating  validatedSeq " << validatedSeq << " lastRotated "
                                   << lastRotated << " deleteInterval " << deleteInterval_
                                   << " canDelete_ " << canDelete_ << " state "
@@ -410,15 +455,16 @@ SHAMapStoreImp::run()
                                   << ledgerMaster_->getValidatedLedgerAge().count()
                                   << "s. Complete ledgers: " << ledgerMaster_->getCompleteLedgers();
 
-            clearPrior(lastRotated);
-            switch (healthWait())
             {
-                case HealthResult::Stopping:
+                RotationPhase phase(*this, ns::phase::clearPrior, lv::clearPrior);
+                clearPrior(lastRotated);
+            }
+            if (auto const r = healthWait(); r != HealthResult::KeepGoing)
+            {
+                outcome.finish(exitFor(r));
+                if (r == HealthResult::Stopping)
                     return;
-                case HealthResult::Expired:
-                    continue;
-                case HealthResult::KeepGoing:
-                    break;
+                continue;
             }
 
             JLOG(journal_.debug()) << "copying ledger " << validatedSeq;
@@ -426,26 +472,27 @@ SHAMapStoreImp::run()
 
             try
             {
+                RotationPhase phase(*this, ns::phase::copy, lv::copy);
                 validatedLedger->stateMap().snapShot(false)->visitNodes(
                     [this, &nodeCount](SHAMapTreeNode const& node) {
                         return copyNode(nodeCount, node);
                     });
+                phase.setAttribute(ns::attr::nodeCount, static_cast<std::int64_t>(nodeCount));
             }
             catch (SHAMapMissingNode const& e)
             {
                 JLOG(journal_.error())
                     << "Missing node while copying ledger before rotate: " << e.what();
+                outcome.finish(ns::RotationExit::MissingNode);
                 continue;
             }
 
-            switch (healthWait())
+            if (auto const r = healthWait(); r != HealthResult::KeepGoing)
             {
-                case HealthResult::Stopping:
+                outcome.finish(exitFor(r));
+                if (r == HealthResult::Stopping)
                     return;
-                case HealthResult::Expired:
-                    continue;
-                case HealthResult::KeepGoing:
-                    break;
+                continue;
             }
             // Only log if we completed without a "health" abort
             JLOG(journal_.debug())
@@ -470,50 +517,59 @@ SHAMapStoreImp::run()
 
             JLOG(journal_.debug()) << "freshening caches";
             freshenCaches();
-            switch (healthWait())
+            if (auto const r = healthWait(); r != HealthResult::KeepGoing)
             {
-                case HealthResult::Stopping:
+                outcome.finish(exitFor(r));
+                if (r == HealthResult::Stopping)
                     return;
-                case HealthResult::Expired:
-                    continue;
-                case HealthResult::KeepGoing:
-                    break;
+                continue;
             }
             // Only log if we completed without a "health" abort
             JLOG(journal_.debug()) << validatedSeq << " freshened caches";
 
             JLOG(journal_.trace()) << "Making a new backend";
-            auto newBackend = makeBackendRotating();
+            auto newBackend = [&] {
+                RotationPhase phase(*this, ns::phase::newBackend, lv::newBackend);
+                return makeBackendRotating();
+            }();
             JLOG(journal_.debug()) << validatedSeq << " new backend " << newBackend->getName();
 
-            clearCaches(validatedSeq);
-            switch (healthWait())
             {
-                case HealthResult::Stopping:
+                RotationPhase phase(*this, ns::phase::clearCaches, lv::clearCaches);
+                clearCaches(validatedSeq);
+            }
+            if (auto const r = healthWait(); r != HealthResult::KeepGoing)
+            {
+                outcome.finish(exitFor(r));
+                if (r == HealthResult::Stopping)
                     return;
-                case HealthResult::Expired:
-                    continue;
-                case HealthResult::KeepGoing:
-                    break;
+                continue;
             }
 
             lastRotated = validatedSeq;
 
-            dbRotating_->rotate(
-                std::move(newBackend),
-                [&](std::string const& writableName, std::string const& archiveName) {
-                    SavedState savedState;
-                    savedState.writableDb = writableName;
-                    savedState.archiveDb = archiveName;
-                    savedState.lastRotated = lastRotated;
-                    stateDb_.setState(savedState);
+            {
+                RotationPhase phase(*this, ns::phase::swap, lv::swap);
+                dbRotating_->rotate(
+                    std::move(newBackend),
+                    [&](std::string const& writableName, std::string const& archiveName) {
+                        SavedState savedState;
+                        savedState.writableDb = writableName;
+                        savedState.archiveDb = archiveName;
+                        savedState.lastRotated = lastRotated;
+                        stateDb_.setState(savedState);
 
-                    clearCaches(validatedSeq);
-                });
+                        clearCaches(validatedSeq);
+                    });
+                phase.setAttribute(
+                    ns::attr::copyForwards,
+                    static_cast<std::int64_t>(dbRotating_->copyForwardTotal()));
+            }
 
             JLOG(journal_.warn()) << "finished rotation. validatedSeq: " << validatedSeq
                                   << ", lastRotated: " << lastRotated
                                   << ". Complete ledgers: " << ledgerMaster_->getCompleteLedgers();
+            outcome.finish(ns::RotationExit::Complete);
         }
     }
 }
@@ -822,8 +878,28 @@ SHAMapStoreImp::healthWait()
         return true;
     };
 
+    std::optional<RotationPhase> waitPhase;
     while (!stop_ && !healthy() && index < circuitBreaker)
     {
+        // Only inside a rotation: the pre-rotation healthWait() at the top of
+        // run() must not open a root span of its own.
+        if (rotating_ && !waitPhase)
+        {
+            waitPhase.emplace(
+                *this,
+                telemetry::nodestore_span::phase::healthWait,
+                telemetry::lval::rotation_phase::healthWait);
+        }
+        if (waitPhase)
+        {
+            waitPhase->setAttribute(
+                telemetry::nodestore_span::attr::serverMode,
+                app_.getOPs().strOperatingMode(mode, false).c_str());
+            waitPhase->setAttribute(
+                telemetry::nodestore_span::attr::missingLedgers,
+                static_cast<std::int64_t>(numMissing));
+        }
+
         // Future-proofing: this value shouldn't change while we are sleeping, but grab it while we
         // have the lock in case it does.
         auto const lowerBound = lastGoodValidatedLedger_;

src/xrpld/app/misc/SHAMapStoreImp.h

@@ -1,8 +1,12 @@
+// cspell:ignore ISTOGRAM Wreturn
 #pragma once
 
 #include <xrpld/app/ledger/LedgerMaster.h>
 #include <xrpld/app/main/Application.h>
 #include <xrpld/app/misc/SHAMapStore.h>
+#include <xrpld/app/misc/SHAMapStoreSpanNames.h>
+#include <xrpld/telemetry/MetricMacros.h>
+#include <xrpld/telemetry/MetricNames.h>
 
 #include <xrpl/beast/utility/Journal.h>
 #include <xrpl/config/BasicConfig.h>
@@ -17,6 +21,7 @@
 #include <xrpl/shamap/FullBelowCache.h>
 #include <xrpl/shamap/SHAMapTreeNode.h>
 #include <xrpl/shamap/TreeNodeCache.h>
+#include <xrpl/telemetry/SpanGuard.h>
 
 #include <algorithm>
 #include <atomic>
@@ -197,13 +202,83 @@ private:
     std::unique_ptr<node_store::Backend>
     makeBackendRotating(std::string path = std::string());
 
+    /**
+     * One rotation phase: a child span of nodestore.rotate for its lifetime
+     * and one rotation_phase_duration_seconds record when it ends. Lives on
+     * the SHAMapStore thread only. Not movable: hold it in a scope.
+     */
+    class RotationPhase
+    {
+    public:
+        template <std::size_t N>
+        RotationPhase(
+            SHAMapStoreImp& owner,
+            telemetry::StaticStr<N> const& phase,
+            char const* stage)
+            : owner_(owner)
+            , stage_(stage)
+            , span_(telemetry::TraceCategory::Ledger, telemetry::nodestore_span::rotateFull, phase)
+        {
+        }
+
+        ~RotationPhase()
+        {
+            auto const seconds =
+                std::chrono::duration<double>(std::chrono::steady_clock::now() - start_).count();
+            XRPL_METRIC_HISTOGRAM_RECORD_LABELED(
+                owner_.app_,
+                telemetry::metric::rotationPhaseDurationSeconds,
+                "Wall-clock seconds spent in one online-delete rotation phase",
+                seconds,
+                {{telemetry::label::stage, std::string(stage_)}});
+        }
+
+        RotationPhase(RotationPhase const&) = delete;
+        RotationPhase&
+        operator=(RotationPhase const&) = delete;
+        RotationPhase(RotationPhase&&) = delete;
+        RotationPhase&
+        operator=(RotationPhase&&) = delete;
+
+        template <class Value>
+        void
+        setAttribute(std::string_view key, Value value) noexcept
+        {
+            span_.setAttribute(key, value);
+        }
+
+    private:
+        SHAMapStoreImp& owner_;
+        char const* stage_;
+        std::chrono::steady_clock::time_point start_ = std::chrono::steady_clock::now();
+        telemetry::ScopedSpanGuard span_;
+    };
+
+    // True while run() is inside its rotation block. Read and written on the
+    // SHAMapStore thread only, so it needs no lock.
+    bool rotating_ = false;
+
     template <class CacheInstance>
     bool
     freshenCache(CacheInstance& cache)
     {
-        std::uint64_t check = 0;
+        namespace ns = telemetry::nodestore_span;
+        namespace lv = telemetry::lval::rotation_phase;
 
-        for (auto const& key : cache.getKeys())
+        // getKeys() copies every key under the cache mutex. It gets its own
+        // phase so the hold is visible on its own, apart from the fetch loop.
+        auto const keys = [&] {
+            RotationPhase phase(*this, ns::phase::freshenKeys, lv::freshenKeys);
+            auto k = cache.getKeys();
+            phase.setAttribute(ns::attr::keyCount, static_cast<std::int64_t>(k.size()));
+            return k;
+        }();
+
+        RotationPhase phase(*this, ns::phase::freshenFetch, lv::freshenFetch);
+        phase.setAttribute(ns::attr::keyCount, static_cast<std::int64_t>(keys.size()));
+
+        std::uint64_t check = 0;
+        for (auto const& key : keys)
         {
             dbRotating_->fetchNodeObject(key, 0, node_store::FetchType::Synchronous, true);
             if (!(++check % checkHealthInterval_) && healthWait() != HealthResult::KeepGoing)