feat(ledger): count acquisition stalls instead of only logging them

A saturated ledgerData lane makes TimeoutCounter re-arm its timer
without running the timer body, so timeouts_ never advances and the
six-timeout give-up can never fire. Acquisitions then neither finish
nor fail until the one-minute sweep destroys their partial maps, and
the work restarts. Every step of that chain was debug-log-only, so a
node at warning level could not be diagnosed after the fact.

The counters are separate on purpose: deferrals rising while timeouts
stay flat is the signature, and no single counter shows it.

Completions are recorded in done() rather than at the "Done: complete"
log line, because that line also fires for failures and misses the
checkLocal and receiveNode paths; done() is the one funnel every
outcome passes through and its signaled_ guard makes it idempotent.

AcquireStats is only forward-declared in ServiceRegistry so libxrpl
still includes nothing from xrpld. The src/ include path for the test
binary moves out of the telemetry guard, since a header-only type
under src/xrpld/ is testable in every build.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

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

@@ -190,6 +190,7 @@ tests.libxrpl > xrpl.basics
 tests.libxrpl > xrpl.config
 tests.libxrpl > xrpl.consensus
 tests.libxrpl > xrpl.core
+tests.libxrpl > xrpld.app
 tests.libxrpl > xrpld.telemetry
 tests.libxrpl > xrpl.json
 tests.libxrpl > xrpl.ledger

include/xrpl/core/ServiceRegistry.h

@@ -48,6 +48,10 @@ using CachedSLEs = TaggedCache<uint256, SLE const>;
 
 // Forward declarations
 class AcceptedLedger;
+// Defined in src/xrpld/app/ledger/AcquireStats.h. Forward-declared here
+// because libxrpl must not include an xrpld header; a reference to an
+// incomplete type is all this interface needs.
+class AcquireStats;
 class AmendmentTable;
 class Cluster;
 class CollectorManager;
@@ -199,6 +203,15 @@ public:
     virtual PendingSaves&
     getPendingSaves() = 0;
 
+    /**
+     * Return the process-wide ledger-acquisition counters.
+     *
+     * Shared by every acquisition, so the counts are process-wide rather than
+     * per-ledger.
+     */
+    virtual AcquireStats&
+    getAcquireStats() = 0;
+
     virtual OpenLedger&
     getOpenLedger() = 0;
 

src/tests/libxrpl/CMakeLists.txt

@@ -21,6 +21,11 @@ set_target_properties(
 )
 # Lets test sources include the shared helpers as <helpers/...>.
 target_include_directories(xrpl_tests PRIVATE ${CMAKE_CURRENT_SOURCE_DIR})
+# Some headers under test live in src/xrpld/ rather than in libxrpl (for
+# example app/ledger/AcquireStats.h and telemetry/ValidationTracker.h), so put
+# src/ on the include path to reach them as <xrpld/...>. This is unconditional
+# because header-only ones are testable in every build, telemetry or not.
+target_include_directories(xrpl_tests PRIVATE ${CMAKE_SOURCE_DIR}/src)
 target_link_libraries(xrpl_tests PRIVATE GTest::gtest GTest::gmock xrpl.libxrpl)
 
 # One source subdirectory per module. Network unit tests are currently not
@@ -30,6 +35,7 @@ set(test_modules
     consensus
     crypto
     json
+    ledger
     peerfinder
     resource
     shamap
@@ -99,9 +105,8 @@ if(telemetry)
             opentelemetry-cpp::opentelemetry-cpp
     )
     # ValidationTracker lives in src/xrpld/ (not libxrpl), so we compile its
-    # implementation directly into the test binary and put src/ on the include
-    # path so its tests can reach <xrpld/telemetry/...> headers.
-    target_include_directories(xrpl_tests PRIVATE ${CMAKE_SOURCE_DIR}/src)
+    # implementation directly into the test binary. The src/ include path its
+    # tests need is added unconditionally above.
     target_sources(
         xrpl_tests
         PRIVATE

src/tests/libxrpl/helpers/TestServiceRegistry.h

@@ -282,6 +282,15 @@ public:
         return pendingSaves_;
     }
 
+    // AcquireStats is declared in src/xrpld/, which this helper deliberately
+    // does not include; returning a reference to an incomplete type is legal
+    // as long as nothing here forms the value.
+    AcquireStats&
+    getAcquireStats() override
+    {
+        throw std::logic_error("TestServiceRegistry::getAcquireStats() not implemented");
+    }
+
     OpenLedger&
     getOpenLedger() override
     {

src/tests/libxrpl/ledger/AcquireStats.cpp

@@ -0,0 +1,199 @@
+/**
+ * @file AcquireStats.cpp
+ * Unit tests for the ledger-acquisition counters (AcquireStats.h).
+ *
+ * The counters exist to make one specific failure readable from metrics: a
+ * saturated job lane makes the acquisition timer re-arm without running its
+ * body, so the retry count never advances and the give-up path never fires.
+ * Deferrals climbing while timeouts stay flat is the only signature of that,
+ * so the tests below pin down that the two counters move independently and
+ * that a stalled run and a healthy run produce different, exact readings.
+ *
+ * AcquireStats is header-only, so nothing from xrpld needs to be linked here.
+ */
+
+#include <xrpld/app/ledger/AcquireStats.h>
+
+#include <gtest/gtest.h>
+
+#include <cstdint>
+#include <thread>
+#include <vector>
+
+namespace {
+
+using xrpl::AcquireStats;
+
+/**
+ * A fresh instance reads zero on every counter, so a later test can attribute
+ * every increment to its own call rather than to construction.
+ */
+TEST(AcquireStatsTest, StartsAtZero)
+{
+    AcquireStats stats;
+
+    EXPECT_EQ(stats.getDeferrals(), 0u);
+    EXPECT_EQ(stats.getTimeouts(), 0u);
+    EXPECT_EQ(stats.getGiveUps(), 0u);
+    EXPECT_EQ(stats.getAborts(), 0u);
+    EXPECT_EQ(stats.getAbortsWithPartialWork(), 0u);
+    EXPECT_EQ(stats.getCompletions(), 0u);
+    EXPECT_EQ(stats.getSweepEvictions(), 0u);
+}
+
+/**
+ * Each counter must move alone, and the others must stay exactly where they
+ * were. A deferral that also nudged the timeout counter would erase the
+ * divergence that identifies the stall, so each step asserts both the counter
+ * that should have moved and the ones that must not have.
+ */
+TEST(AcquireStatsTest, CountersAdvanceIndependently)
+{
+    AcquireStats stats;
+
+    stats.recordDeferral();
+    stats.recordDeferral();
+    stats.recordDeferral();
+    EXPECT_EQ(stats.getDeferrals(), 3u);
+    EXPECT_EQ(stats.getTimeouts(), 0u);
+    EXPECT_EQ(stats.getGiveUps(), 0u);
+
+    stats.recordTimeout();
+    EXPECT_EQ(stats.getTimeouts(), 1u);
+    EXPECT_EQ(stats.getDeferrals(), 3u);
+
+    stats.recordGiveUp();
+    EXPECT_EQ(stats.getGiveUps(), 1u);
+    EXPECT_EQ(stats.getTimeouts(), 1u);
+    EXPECT_EQ(stats.getDeferrals(), 3u);
+
+    stats.recordCompletion();
+    stats.recordCompletion();
+    EXPECT_EQ(stats.getCompletions(), 2u);
+    EXPECT_EQ(stats.getSweepEvictions(), 0u);
+
+    stats.recordSweepEviction();
+    EXPECT_EQ(stats.getSweepEvictions(), 1u);
+    EXPECT_EQ(stats.getCompletions(), 2u);
+
+    // Nothing above records an abort, so both abort counters stay at zero.
+    EXPECT_EQ(stats.getAborts(), 0u);
+    EXPECT_EQ(stats.getAbortsWithPartialWork(), 0u);
+}
+
+/**
+ * An abort that discards a partly built map is the expensive case, because the
+ * whole acquisition restarts. It must be distinguishable from a cheap abort
+ * that had built nothing yet, and the cheap one must leave the partial-work
+ * counter untouched.
+ */
+TEST(AcquireStatsTest, AbortDistinguishesPartialWork)
+{
+    AcquireStats stats;
+
+    stats.recordAbort(false);
+    EXPECT_EQ(stats.getAborts(), 1u);
+    EXPECT_EQ(stats.getAbortsWithPartialWork(), 0u);
+
+    stats.recordAbort(true);
+    EXPECT_EQ(stats.getAborts(), 2u);
+    EXPECT_EQ(stats.getAbortsWithPartialWork(), 1u);
+
+    // An abort never counts as a completion or a give-up.
+    EXPECT_EQ(stats.getCompletions(), 0u);
+    EXPECT_EQ(stats.getGiveUps(), 0u);
+}
+
+/**
+ * The stalled shape: many deferrals, no timeouts, no completions, and sweeps
+ * discarding partial work. Timeouts staying at exactly zero is what proves the
+ * give-up path cannot fire, so that assertion is the point of the test.
+ */
+TEST(AcquireStatsTest, StalledShapeIsDistinguishable)
+{
+    AcquireStats stalled;
+    for (int i = 0; i < 1000; ++i)
+        stalled.recordDeferral();
+    for (int i = 0; i < 5; ++i)
+    {
+        stalled.recordSweepEviction();
+        stalled.recordAbort(true);
+    }
+
+    EXPECT_EQ(stalled.getDeferrals(), 1000u);
+    EXPECT_EQ(stalled.getTimeouts(), 0u);
+    EXPECT_EQ(stalled.getGiveUps(), 0u);
+    EXPECT_EQ(stalled.getCompletions(), 0u);
+    EXPECT_EQ(stalled.getSweepEvictions(), 5u);
+    EXPECT_EQ(stalled.getAborts(), 5u);
+    EXPECT_EQ(stalled.getAbortsWithPartialWork(), 5u);
+}
+
+/**
+ * The healthy shape, for contrast: timeouts accrue, give-up fires, and
+ * completions dominate. The same seven counters, read the opposite way, which
+ * is what makes the stalled reading above meaningful.
+ */
+TEST(AcquireStatsTest, HealthyShapeIsDistinguishable)
+{
+    AcquireStats healthy;
+    for (int i = 0; i < 10; ++i)
+        healthy.recordDeferral();
+    for (int i = 0; i < 7; ++i)
+        healthy.recordTimeout();
+    healthy.recordGiveUp();
+    for (int i = 0; i < 27; ++i)
+        healthy.recordCompletion();
+
+    EXPECT_EQ(healthy.getDeferrals(), 10u);
+    EXPECT_EQ(healthy.getTimeouts(), 7u);
+    EXPECT_EQ(healthy.getGiveUps(), 1u);
+    EXPECT_EQ(healthy.getCompletions(), 27u);
+    EXPECT_EQ(healthy.getSweepEvictions(), 0u);
+    EXPECT_EQ(healthy.getAborts(), 0u);
+    EXPECT_EQ(healthy.getAbortsWithPartialWork(), 0u);
+}
+
+/**
+ * Concurrent recording must lose no increment. Every counter is read as a
+ * rate, so a dropped increment silently reads as a slower rate rather than as
+ * an error.
+ *
+ * The expected totals below are written as literals on purpose. Deriving them
+ * from the loop bounds would make the assertion agree with the loop by
+ * construction and pass even if every increment were lost.
+ */
+TEST(AcquireStatsTest, ConcurrentRecordingLosesNothing)
+{
+    AcquireStats stats;
+    constexpr int kThreads = 4;
+    constexpr int kPerThread = 1000;
+
+    std::vector<std::thread> threads;
+    threads.reserve(kThreads);
+    for (int t = 0; t < kThreads; ++t)
+    {
+        threads.emplace_back([&stats]() {
+            for (int i = 0; i < kPerThread; ++i)
+            {
+                stats.recordDeferral();
+                stats.recordCompletion();
+            }
+        });
+    }
+    for (auto& th : threads)
+        th.join();
+
+    // 4 threads * 1000 iterations, stated independently of the loop bounds.
+    EXPECT_EQ(stats.getDeferrals(), std::uint64_t{4000});
+    EXPECT_EQ(stats.getCompletions(), std::uint64_t{4000});
+
+    // The threads record only those two events, so the rest stay at zero.
+    EXPECT_EQ(stats.getTimeouts(), 0u);
+    EXPECT_EQ(stats.getGiveUps(), 0u);
+    EXPECT_EQ(stats.getAborts(), 0u);
+    EXPECT_EQ(stats.getAbortsWithPartialWork(), 0u);
+    EXPECT_EQ(stats.getSweepEvictions(), 0u);
+}
+
+}  // namespace

src/tests/libxrpl/telemetry/MetricsRegistry.cpp

@@ -475,6 +475,13 @@ public:
     {
         throwUnimplemented();
     }
+    // AcquireStats lives in src/xrpld/ and is only forward-declared here; a
+    // reference return to an incomplete type is fine because this throws.
+    AcquireStats&
+    getAcquireStats() override
+    {
+        throwUnimplemented();
+    }
     [[nodiscard]] OpenLedger&
     getOpenLedger() override
     {

src/xrpld/app/ledger/AcquireStats.h

@@ -0,0 +1,258 @@
+#pragma once
+
+#include <atomic>
+#include <cstdint>
+
+namespace xrpl {
+
+/**
+ * Counters for ledger-acquisition progress and stalls.
+ *
+ * Each event counted here is otherwise visible only as a debug log line, so a
+ * node running at warning level cannot be diagnosed after the fact. The
+ * counters exist so the acquisition state machine can be observed rather than
+ * inferred.
+ *
+ * The diagnostic value is in the pairs, not in the individual counts:
+ *
+ * ```
+ *   deferrals rising + timeouts flat  -> give-up path disarmed
+ *   sweeps rising + completions zero  -> partial work discarded, then redone
+ *   completions rising                -> healthy, whatever else is high
+ * ```
+ *
+ * A deferral happens when the job lane is already at its limit. The timer is
+ * re-armed without running the timer body, so the timeout that would
+ * eventually trigger give-up never accrues. That divergence is invisible
+ * unless deferrals and timeouts are counted separately, which is why they are
+ * two counters and not one.
+ *
+ * Who writes each counter:
+ *
+ * ```
+ *   TimeoutCounter ---- recordDeferral() ------> +-----------------+
+ *                  \--- recordTimeout() ------->|                 |
+ *                                               |                 |
+ *   InboundLedger  ---- recordGiveUp() -------->|  AcquireStats   |
+ *                  |--- recordCompletion() ---->|   (7 atomic     |
+ *                  \--- recordAbort() --------->|    counters)    |
+ *                                               |                 |
+ *   InboundLedgers ---- recordSweepEviction() ->+-----------------+
+ *                                                       |
+ *                                        get*() reads   v
+ *                                               metrics exporter
+ * ```
+ *
+ * A single instance is owned by the application and shared by all
+ * acquisitions, so the counts are process-wide rather than per-ledger.
+ *
+ * Example - detect the stall:
+ * @code
+ * auto const deferred = stats.getDeferrals() - prevDeferrals;
+ * auto const timedOut = stats.getTimeouts() - prevTimeouts;
+ * if (deferred > 100 && timedOut == 0)
+ * {
+ *     // The give-up path cannot fire, so acquisitions will never end.
+ * }
+ * @endcode
+ *
+ * Example - separate cheap restarts from expensive ones:
+ * @code
+ * auto const wasted = stats.getAbortsWithPartialWork() - prevAbortsWithWork;
+ * if (wasted > 0)
+ * {
+ *     // Partly built maps were thrown away; that work has to be redone.
+ * }
+ * @endcode
+ *
+ * Example - edge case, a quiet node:
+ * @code
+ * // Every delta being zero means no acquisition was attempted, which is not
+ * // the same as healthy. Check completions before concluding anything.
+ * @endcode
+ *
+ * @note Thread-safe. Every counter is an independent relaxed atomic, so any
+ *       number of threads may record concurrently without losing an
+ *       increment. Because the counters are independent, no read across two
+ *       of them is a consistent snapshot. Compare rates over an interval
+ *       rather than instantaneous values.
+ * @note All counters are monotonic for the life of the process and are never
+ *       reset, so a reader differences successive samples to get a rate. They
+ *       saturate only on 64-bit wraparound, which is unreachable in practice.
+ * @note Completions count acquisitions that ended successfully, whether the
+ *       data came from peers or was already present locally. They do not
+ *       count an acquisition that is still in flight.
+ */
+class AcquireStats
+{
+public:
+    /**
+     * Record that a timer job was skipped because its job lane was full.
+     *
+     * The timer is re-armed but its body does not run, so this does not
+     * advance the retry count toward give-up.
+     */
+    void
+    recordDeferral()
+    {
+        deferrals_.fetch_add(1, std::memory_order_relaxed);
+    }
+
+    /**
+     * Record that a timer body ran and advanced the retry count.
+     *
+     * This is the counter that give-up is measured against, so it is kept
+     * separate from deferrals.
+     */
+    void
+    recordTimeout()
+    {
+        timeouts_.fetch_add(1, std::memory_order_relaxed);
+    }
+
+    /**
+     * Record that an acquisition exceeded its retry budget and failed
+     * cleanly.
+     */
+    void
+    recordGiveUp()
+    {
+        giveUps_.fetch_add(1, std::memory_order_relaxed);
+    }
+
+    /**
+     * Record that an acquisition was destroyed before finishing.
+     *
+     * @param hadPartialWork True if part of a map had already been built.
+     *                       That work is discarded and has to be redone, so
+     *                       it is counted separately as the expensive case.
+     */
+    void
+    recordAbort(bool hadPartialWork)
+    {
+        aborts_.fetch_add(1, std::memory_order_relaxed);
+        if (hadPartialWork)
+            abortsWithPartialWork_.fetch_add(1, std::memory_order_relaxed);
+    }
+
+    /**
+     * Record that an acquisition finished successfully.
+     */
+    void
+    recordCompletion()
+    {
+        completions_.fetch_add(1, std::memory_order_relaxed);
+    }
+
+    /**
+     * Record that an idle acquisition was evicted by the sweep.
+     */
+    void
+    recordSweepEviction()
+    {
+        sweepEvictions_.fetch_add(1, std::memory_order_relaxed);
+    }
+
+    /**
+     * Return the number of timer jobs skipped because a lane was full.
+     */
+    [[nodiscard]] std::uint64_t
+    getDeferrals() const
+    {
+        return deferrals_.load(std::memory_order_relaxed);
+    }
+
+    /**
+     * Return the number of timer bodies that ran without progress.
+     */
+    [[nodiscard]] std::uint64_t
+    getTimeouts() const
+    {
+        return timeouts_.load(std::memory_order_relaxed);
+    }
+
+    /**
+     * Return the number of acquisitions that exhausted their retry budget.
+     */
+    [[nodiscard]] std::uint64_t
+    getGiveUps() const
+    {
+        return giveUps_.load(std::memory_order_relaxed);
+    }
+
+    /**
+     * Return the number of acquisitions destroyed before finishing.
+     */
+    [[nodiscard]] std::uint64_t
+    getAborts() const
+    {
+        return aborts_.load(std::memory_order_relaxed);
+    }
+
+    /**
+     * Return the number of aborts that discarded a partly built map.
+     *
+     * This is a subset of getAborts().
+     */
+    [[nodiscard]] std::uint64_t
+    getAbortsWithPartialWork() const
+    {
+        return abortsWithPartialWork_.load(std::memory_order_relaxed);
+    }
+
+    /**
+     * Return the number of acquisitions that finished successfully.
+     */
+    [[nodiscard]] std::uint64_t
+    getCompletions() const
+    {
+        return completions_.load(std::memory_order_relaxed);
+    }
+
+    /**
+     * Return the number of idle acquisitions evicted by the sweep.
+     */
+    [[nodiscard]] std::uint64_t
+    getSweepEvictions() const
+    {
+        return sweepEvictions_.load(std::memory_order_relaxed);
+    }
+
+private:
+    /**
+     * Timer jobs skipped because the job lane was at its limit.
+     */
+    std::atomic<std::uint64_t> deferrals_{0};
+
+    /**
+     * Timer bodies that ran and advanced the retry count.
+     */
+    std::atomic<std::uint64_t> timeouts_{0};
+
+    /**
+     * Acquisitions that exhausted their retry budget.
+     */
+    std::atomic<std::uint64_t> giveUps_{0};
+
+    /**
+     * Acquisitions destroyed before finishing.
+     */
+    std::atomic<std::uint64_t> aborts_{0};
+
+    /**
+     * Aborts that discarded a partly built map.
+     */
+    std::atomic<std::uint64_t> abortsWithPartialWork_{0};
+
+    /**
+     * Acquisitions that finished successfully.
+     */
+    std::atomic<std::uint64_t> completions_{0};
+
+    /**
+     * Idle acquisitions evicted by the sweep.
+     */
+    std::atomic<std::uint64_t> sweepEvictions_{0};
+};
+
+}  // namespace xrpl

src/xrpld/app/ledger/detail/InboundLedger.cpp

@@ -1,6 +1,7 @@
 #include <xrpld/app/ledger/InboundLedger.h>
 
 #include <xrpld/app/ledger/AccountStateSF.h>
+#include <xrpld/app/ledger/AcquireStats.h>
 #include <xrpld/app/ledger/InboundLedgers.h>
 #include <xrpld/app/ledger/LedgerMaster.h>
 #include <xrpld/app/ledger/TransactionStateSF.h>
@@ -212,6 +213,10 @@ InboundLedger::~InboundLedger()
     }
     if (!isDone())
     {
+        // Partial work means a map was partly built and is now discarded, so
+        // the whole acquisition has to start over. That is the expensive case,
+        // so it is counted apart from a cheap abort that had nothing yet.
+        app_.getAcquireStats().recordAbort(haveHeader_ || haveState_ || haveTransactions_);
         JLOG(journal_.debug()) << "Acquire " << hash_ << " abort "
                                << ((timeouts_ == 0) ? std::string()
                                                     : (std::string("timeouts:") +
@@ -386,6 +391,7 @@ InboundLedger::onTimer(bool wasProgress, ScopedLockType&)
 
     if (timeouts_ > kLedgerTimeoutRetriesMax)
     {
+        app_.getAcquireStats().recordGiveUp();
         if (seq_ != 0)
         {
             JLOG(journal_.warn()) << timeouts_ << " timeouts for ledger " << seq_;
@@ -452,6 +458,13 @@ InboundLedger::done()
     signaled_ = true;
     touch();
 
+    // Counted here rather than at any single caller because done() is the one
+    // funnel every peer-driven outcome passes through, and the signaled_ guard
+    // above makes it run at most once per acquisition. failed_ outcomes are
+    // excluded; the give-up path counts those itself.
+    if (complete_ && !failed_)
+        app_.getAcquireStats().recordCompletion();
+
     // Finalize the acquire span with the outcome, timeout count, and peer
     // count. Keep it active as the ambient context across the finalize and
     // the outcome log below so that log line carries the span's trace_id.

src/xrpld/app/ledger/detail/InboundLedgers.cpp

@@ -1,5 +1,6 @@
 #include <xrpld/app/ledger/InboundLedgers.h>
 
+#include <xrpld/app/ledger/AcquireStats.h>
 #include <xrpld/app/ledger/InboundLedger.h>
 #include <xrpld/app/ledger/LedgerMaster.h>
 #include <xrpld/app/main/Application.h>
@@ -399,6 +400,10 @@ public:
                 else if ((la + std::chrono::minutes(1)) < start)
                 {
                     stuffToSweep.push_back(it->second);
+                    // An eviction here discards whatever the acquisition had
+                    // built, so the work restarts. Counted to tell that apart
+                    // from an acquisition that ended on its own.
+                    app_.getAcquireStats().recordSweepEviction();
                     // shouldn't cause the actual final delete
                     // since we are holding a reference in the vector.
                     it = ledgers_.erase(it);

src/xrpld/app/ledger/detail/TimeoutCounter.cpp

@@ -1,5 +1,6 @@
 #include <xrpld/app/ledger/detail/TimeoutCounter.h>
 
+#include <xrpld/app/ledger/AcquireStats.h>
 #include <xrpld/app/main/Application.h>
 
 #include <xrpl/basics/Log.h>
@@ -63,6 +64,10 @@ TimeoutCounter::queueJob(ScopedLockType& sl)
         app_.getJobQueue().getJobCountTotal(queueJobParameter_.jobType) >=
             queueJobParameter_.jobLimit)
     {
+        // Counted separately from timeouts: this path re-arms the timer
+        // without running invokeOnTimer, so timeouts_ does not advance and the
+        // give-up test that reads it cannot fire while the lane stays full.
+        app_.getAcquireStats().recordDeferral();
         JLOG(journal_.debug()) << "Deferring " << queueJobParameter_.jobName
                                << " timer due to load";
         setTimer(sl);
@@ -87,6 +92,7 @@ TimeoutCounter::invokeOnTimer()
     if (!progress_)
     {
         ++timeouts_;
+        app_.getAcquireStats().recordTimeout();
         JLOG(journal_.debug()) << "Timeout(" << timeouts_ << ") "
                                << " acquiring " << hash_;
         onTimer(false, sl);

src/xrpld/app/main/Application.cpp

@@ -1,6 +1,7 @@
 #include <xrpld/app/main/Application.h>
 
 #include <xrpld/app/consensus/RCLValidations.h>
+#include <xrpld/app/ledger/AcquireStats.h>
 #include <xrpld/app/ledger/InboundLedger.h>
 #include <xrpld/app/ledger/InboundLedgers.h>
 #include <xrpld/app/ledger/InboundTransactions.h>
@@ -236,6 +237,11 @@ public:
     NodeStoreScheduler nodeStoreScheduler_;
     std::unique_ptr<SHAMapStore> shaMapStore_;
     PendingSaves pendingSaves_;
+    /**
+     * Process-wide ledger-acquisition counters, shared by every acquisition.
+     * Declared before the ledger services below so it outlives them.
+     */
+    AcquireStats acquireStats_;
     std::optional<OpenLedger> openLedger_;
 
     NodeCache tempNodeCache_;
@@ -816,6 +822,12 @@ public:
         return pendingSaves_;
     }
 
+    AcquireStats&
+    getAcquireStats() override
+    {
+        return acquireStats_;
+    }
+
     OpenLedger&
     getOpenLedger() override
     {