feat(basics): expose the peak TaggedCache lock hold and warn on a one-second hold

Preparing to point the finger at TaggedCache's mutex when the whole process
freezes at rotation time. Every `sweep()` and `getKeys()` calls `noteLockHold`
after releasing the mutex; `takeLockHoldPeak()` returns the longest hold since
the last call and resets. A one-second hold logs a warn line naming the op
and the entry count — the same one-second bar `LoadMonitor::addLoadSample`
uses to flag a job.

`lockHoldPeakNs_` is a mutable atomic so `getKeys() const` can note its own
hold; `FullBelowCache` forwards `takeLockHoldPeak()` so the metrics layer can
read either cache the same way. Nothing here depends on telemetry — this file
lives in `xrpl/basics` and must not.

Test pins the behaviour end to end: neither getKeys() nor sweep() records
anything on an empty cache; both push the peak above zero once the cache has
200,000 entries; and takeLockHoldPeak() is destructive.

include/xrpl/basics/TaggedCache.h

@@ -101,6 +101,14 @@ public:
     int
     getTrackSize() const;
 
+    /**
+     * Longest single hold of the cache mutex by sweep() or getKeys() since
+     * the previous call, then reset to zero. A per-collect peak: the metrics
+     * gauge reads it once per collection tick.
+     */
+    [[nodiscard]] std::chrono::nanoseconds
+    takeLockHoldPeak() noexcept;
+
     float
     getHitRate();
 
@@ -375,6 +383,18 @@ private:
         std::atomic<int>& allRemovals,
         std::scoped_lock<std::recursive_mutex> const&);
 
+    /**
+     * Record one mutex hold. Keeps the maximum since the last take and warns
+     * when a hold reaches one second, the same bar LoadMonitor uses for a job.
+     * `const` because getKeys() is `const` and lockHoldPeakNs_ is `mutable`.
+     */
+    void
+    noteLockHold(std::chrono::steady_clock::time_point start, std::size_t entries, char const* op)
+        const noexcept;
+
+    // Peak mutex hold in nanoseconds since the last takeLockHoldPeak().
+    mutable std::atomic<std::int64_t> lockHoldPeakNs_{0};
+
     beast::Journal journal_;
     clock_type& clock_;
     Stats stats_;

include/xrpl/basics/TaggedCache.ipp

@@ -137,6 +137,53 @@ TaggedCache<Key, T, IsKeyCache, SharedWeakUnionPointer, SharedPointerType, Hash,
     return cache_.size();
 }
 
+template <
+    class Key,
+    class T,
+    bool IsKeyCache,
+    class SharedWeakUnionPointer,
+    class SharedPointerType,
+    class Hash,
+    class KeyEqual,
+    class Mutex>
+inline std::chrono::nanoseconds
+TaggedCache<Key, T, IsKeyCache, SharedWeakUnionPointer, SharedPointerType, Hash, KeyEqual, Mutex>::
+    takeLockHoldPeak() noexcept
+{
+    return std::chrono::nanoseconds{lockHoldPeakNs_.exchange(0, std::memory_order_relaxed)};
+}
+
+template <
+    class Key,
+    class T,
+    bool IsKeyCache,
+    class SharedWeakUnionPointer,
+    class SharedPointerType,
+    class Hash,
+    class KeyEqual,
+    class Mutex>
+inline void
+TaggedCache<Key, T, IsKeyCache, SharedWeakUnionPointer, SharedPointerType, Hash, KeyEqual, Mutex>::
+    noteLockHold(std::chrono::steady_clock::time_point start, std::size_t entries, char const* op)
+        const noexcept
+{
+    using namespace std::chrono;
+    auto const held = steady_clock::now() - start;
+    auto const heldNs = duration_cast<nanoseconds>(held).count();
+    // fetch_max is C++26; a CAS loop is the portable maximum.
+    auto seen = lockHoldPeakNs_.load(std::memory_order_relaxed);
+    while (seen < heldNs &&
+           !lockHoldPeakNs_.compare_exchange_weak(seen, heldNs, std::memory_order_relaxed))
+    {
+    }
+    if (held >= seconds{1})
+    {
+        JLOG(journal_.warn()) << name_ << " TaggedCache " << op << " held the lock "
+                              << duration_cast<milliseconds>(held).count() << "ms over " << entries
+                              << " entries";
+    }
+}
+
 template <
     class Key,
     class T,
@@ -241,8 +288,10 @@ TaggedCache<Key, T, IsKeyCache, SharedWeakUnionPointer, SharedPointerType, Hash,
     clock_type::time_point whenExpire;
 
     auto const start = std::chrono::steady_clock::now();
+    std::size_t entries = 0;
     {
         std::scoped_lock const lock(mutex_);
+        entries = cache_.size();
 
         if (targetSize_ == 0 || (static_cast<int>(cache_.size()) <= targetSize_))
         {
@@ -277,6 +326,7 @@ TaggedCache<Key, T, IsKeyCache, SharedWeakUnionPointer, SharedPointerType, Hash,
     }
     // At this point allStuffToSweep will go out of scope outside the lock
     // and decrement the reference count on each strong pointer.
+    noteLockHold(start, entries, "sweep");
     JLOG(journal_.debug()) << name_ << " TaggedCache sweep lock duration "
                            << std::chrono::duration_cast<std::chrono::milliseconds>(
                                   std::chrono::steady_clock::now() - start)
@@ -640,8 +690,10 @@ TaggedCache<Key, T, IsKeyCache, SharedWeakUnionPointer, SharedPointerType, Hash,
         XRPL_ASSERT(lock.owns_lock(), "xrpl::TaggedCache::getKeys(): owns lock");
         XRPL_ASSERT(
             v.capacity() >= cache_.size(), "xrpl::TaggedCache::getKeys(): sufficient capacity");
+        auto const copyStart = std::chrono::steady_clock::now();
         for (auto const& _ : cache_)
             v.push_back(_.first);
+        noteLockHold(copyStart, v.size(), "getKeys");
     }
 
     return v;

include/xrpl/shamap/FullBelowCache.h

@@ -82,6 +82,16 @@ public:
         cache_.sweep();
     }
 
+    /**
+     * See TaggedCache::takeLockHoldPeak(). Longest mutex hold since the last
+     * call, then reset. Read once per metrics collection tick.
+     */
+    [[nodiscard]] std::chrono::nanoseconds
+    takeLockHoldPeak() noexcept
+    {
+        return cache_.takeLockHoldPeak();
+    }
+
     /**
      * Refresh the last access time of an item, if it exists.
      * Thread safety:

src/tests/libxrpl/basics/TaggedCache.cpp

@@ -243,4 +243,36 @@ TEST(TaggedCacheTest, tagged_cache)
     }
 }
 
+TEST(TaggedCacheTest, lock_hold_peak_records_getkeys_and_sweep_then_resets)
+{
+    using namespace std::chrono_literals;
+    beast::Journal const journal{TestSink::instance()};
+    TestStopwatch clock;
+    clock.set(0);
+
+    using Cache = TaggedCache<LedgerIndex, std::string>;
+    Cache c("peak", 0, 1s, clock, journal);
+
+    // Nothing has held the lock yet.
+    EXPECT_EQ(c.takeLockHoldPeak(), 0ns);
+
+    // Enough entries that copying every key takes a measurable time.
+    for (LedgerIndex i = 0; i < 200'000; ++i)
+        c.insert(i, "v");
+
+    auto const keys = c.getKeys();
+    ASSERT_EQ(keys.size(), 200'000u);
+    auto const afterGetKeys = c.takeLockHoldPeak();
+    EXPECT_GT(afterGetKeys, 0ns);
+    // take() is destructive: the next read starts from zero.
+    EXPECT_EQ(c.takeLockHoldPeak(), 0ns);
+
+    // A sweep that expires everything also holds the lock over every entry.
+    ++clock;
+    ++clock;
+    c.sweep();
+    EXPECT_GT(c.takeLockHoldPeak(), 0ns);
+    EXPECT_EQ(c.getTrackSize(), 0);
+}
+
 }  // namespace xrpl