Merge branch 'develop' into vlntb/mem-leak-ledger-history-3

include/xrpl/basics/TaggedCache.h

@@ -143,9 +143,6 @@ public:
     bool
     retrieve(key_type const& key, T& data);
 
-    mutex_type&
-    peekMutex();
-
     std::vector<key_type>
     getKeys() const;
 
@@ -164,6 +161,21 @@ public:
     fetch(key_type const& digest, Handler const& h);
     // End CachedSLEs functions.
 
+    /** Fetch or create an entry and execute a callback while holding the lock.
+
+        The entry for the given key is fetched from the cache or created if it
+        doesn't exist. The callback is then invoked with a reference to the
+        entry while the cache mutex is still held, allowing safe modification
+        of the cached object.
+
+        @param key The key to fetch or create
+        @param callback Function to call with the entry under lock.
+                        Signature: void(T&)
+    */
+    template <class Callback>
+    void
+    fetch_and_modify(key_type const& key, Callback&& callback);
+
 private:
     SharedPointerType
     initialFetch(key_type const& key, std::lock_guard<mutex_type> const& l);

include/xrpl/basics/TaggedCache.ipp

@@ -508,22 +508,6 @@ TaggedCache<Key, T, IsKeyCache, SharedWeakUnionPointer, SharedPointerType, Hash,
     return true;
 }
 
-template <
-    class Key,
-    class T,
-    bool IsKeyCache,
-    class SharedWeakUnionPointer,
-    class SharedPointerType,
-    class Hash,
-    class KeyEqual,
-    class Mutex>
-inline auto
-TaggedCache<Key, T, IsKeyCache, SharedWeakUnionPointer, SharedPointerType, Hash, KeyEqual, Mutex>::
-    peekMutex() -> mutex_type&
-{
-    return m_mutex;
-}
-
 template <
     class Key,
     class T,
@@ -602,6 +586,31 @@ TaggedCache<Key, T, IsKeyCache, SharedWeakUnionPointer, SharedPointerType, Hash,
 }
 // End CachedSLEs functions.
 
+template <
+    class Key,
+    class T,
+    bool IsKeyCache,
+    class SharedWeakUnionPointer,
+    class SharedPointerType,
+    class Hash,
+    class KeyEqual,
+    class Mutex>
+template <class Callback>
+inline void
+TaggedCache<Key, T, IsKeyCache, SharedWeakUnionPointer, SharedPointerType, Hash, KeyEqual, Mutex>::
+    fetch_and_modify(key_type const& key, Callback&& callback)
+{
+    static_assert(
+        !IsKeyCache, "fetch_and_modify is only supported for value caches, not key-only caches");
+
+    std::lock_guard lock(m_mutex);
+
+    auto entry = std::make_shared<T>();
+    canonicalize(key, entry, []() { return false; });
+
+    callback(*entry);
+}
+
 template <
     class Key,
     class T,

src/test/app/LedgerHistory_test.cpp

@@ -67,6 +67,67 @@ public:
         return res;
     }
 
+    void
+    testHashIndexInvariant()
+    {
+        testcase("LedgerHistory hash/index invariant");
+        using namespace jtx;
+        using namespace std::chrono;
+
+        Env env{*this};
+        LedgerHistory lh{beast::insight::NullCollector::New(), env.app()};
+
+        // Create and insert validated ledgers
+        auto const genesis = makeLedger({}, env, lh, 0s);
+        auto const ledger1 = makeLedger(genesis, env, lh, 4s);
+        auto const ledger2 = makeLedger(ledger1, env, lh, 4s);
+        auto const ledger3 = makeLedger(ledger2, env, lh, 4s);
+
+        // Insert as validated (so they go into by_index)
+        lh.insert(genesis, true);
+        lh.insert(ledger1, true);
+        lh.insert(ledger2, true);
+        lh.insert(ledger3, true);
+
+        // Verify the hash/index invariant holds
+        // Can retrieve by sequence and get correct hash
+        BEAST_EXPECT(lh.getLedgerHash(genesis->header().seq) == genesis->header().hash);
+        BEAST_EXPECT(lh.getLedgerHash(ledger1->header().seq) == ledger1->header().hash);
+        BEAST_EXPECT(lh.getLedgerHash(ledger2->header().seq) == ledger2->header().hash);
+        BEAST_EXPECT(lh.getLedgerHash(ledger3->header().seq) == ledger3->header().hash);
+
+        // Can retrieve by sequence and get correct ledger
+        auto fetched1 = lh.getLedgerBySeq(ledger1->header().seq);
+        BEAST_EXPECT(fetched1 != nullptr);
+        BEAST_EXPECT(fetched1->header().hash == ledger1->header().hash);
+
+        auto fetched2 = lh.getLedgerBySeq(ledger2->header().seq);
+        BEAST_EXPECT(fetched2 != nullptr);
+        BEAST_EXPECT(fetched2->header().hash == ledger2->header().hash);
+
+        // Clear ledgers prior to ledger2's sequence
+        lh.clearLedgerCachePrior(ledger2->header().seq);
+
+        // Verify old entries are gone from the in-memory by_index map
+        // Note: getLedgerHash checks by_index directly without DB fallback
+        BEAST_EXPECT(lh.getLedgerHash(genesis->header().seq).isZero());
+        BEAST_EXPECT(lh.getLedgerHash(ledger1->header().seq).isZero());
+
+        // Verify newer entries are still present in by_index
+        BEAST_EXPECT(lh.getLedgerHash(ledger2->header().seq) == ledger2->header().hash);
+        BEAST_EXPECT(lh.getLedgerHash(ledger3->header().seq) == ledger3->header().hash);
+
+        // Verify newer entries remain retrievable and consistent
+        // getLedgerBySeq uses by_index first, then falls back to DB if needed
+        auto fetched2After = lh.getLedgerBySeq(ledger2->header().seq);
+        BEAST_EXPECT(fetched2After != nullptr);
+        BEAST_EXPECT(fetched2After->header().hash == ledger2->header().hash);
+
+        auto fetched3After = lh.getLedgerBySeq(ledger3->header().seq);
+        BEAST_EXPECT(fetched3After != nullptr);
+        BEAST_EXPECT(fetched3After->header().hash == ledger3->header().hash);
+    }
+
     void
     testHandleMismatch()
     {
@@ -163,6 +224,7 @@ public:
     void
     run() override
     {
+        testHashIndexInvariant();
         testHandleMismatch();
     }
 };

src/test/basics/TaggedCache_test.cpp

@@ -20,6 +20,13 @@ original object.
 class TaggedCache_test : public beast::unit_test::suite
 {
 public:
+    // Mutable value type for testing fetch_and_modify
+    struct MutableValue
+    {
+        int counter = 0;
+        std::string name;
+    };
+
     void
     run() override
     {
@@ -129,6 +136,54 @@ public:
             BEAST_EXPECT(c.getCacheSize() == 0);
             BEAST_EXPECT(c.getTrackSize() == 0);
         }
+
+        // Test fetch_and_modify: insert on miss, modify on hit
+        {
+            using MutCache = TaggedCache<Key, MutableValue>;
+            MutCache mc("mutable_test", 2, 2s, clock, journal);
+
+            // A. Insert on miss: fetch_and_modify creates entry and mutates it
+            mc.fetch_and_modify(5, [](MutableValue& v) {
+                v.counter = 42;
+                v.name = "initial";
+            });
+
+            BEAST_EXPECT(mc.getCacheSize() == 1);
+            BEAST_EXPECT(mc.getTrackSize() == 1);
+
+            // Verify the mutation persisted
+            auto p1 = mc.fetch(5);
+            BEAST_EXPECT(p1 != nullptr);
+            BEAST_EXPECT(p1->counter == 42);
+            BEAST_EXPECT(p1->name == "initial");
+
+            // B. Modify existing object on hit
+            // Keep strong pointer to verify in-place modification
+            auto p2 = mc.fetch(5);
+            BEAST_EXPECT(p2 != nullptr);
+            BEAST_EXPECT(p1.get() == p2.get());  // Same object
+
+            // Modify through fetch_and_modify
+            mc.fetch_and_modify(5, [](MutableValue& v) {
+                v.counter += 10;
+                v.name = "modified";
+            });
+
+            // Verify no new entry was created
+            BEAST_EXPECT(mc.getCacheSize() == 1);
+            BEAST_EXPECT(mc.getTrackSize() == 1);
+
+            // Verify the same object was mutated (strong pointer sees change)
+            BEAST_EXPECT(p1->counter == 52);
+            BEAST_EXPECT(p1->name == "modified");
+            BEAST_EXPECT(p2->counter == 52);  // Original pointer sees mutation
+
+            // Verify via fresh fetch
+            auto p3 = mc.fetch(5);
+            BEAST_EXPECT(p3 != nullptr);
+            BEAST_EXPECT(p3.get() == p1.get());  // Same object identity
+            BEAST_EXPECT(p3->counter == 52);
+        }
     }
 };
 

src/xrpld/app/ledger/LedgerHistory.cpp

@@ -8,18 +8,10 @@
 
 namespace xrpl {
 
-// FIXME: Need to clean up ledgers by index at some point
-
 LedgerHistory::LedgerHistory(beast::insight::Collector::ptr const& collector, Application& app)
     : app_(app)
     , collector_(collector)
     , mismatch_counter_(collector->make_counter("ledger.history", "mismatch"))
-    , m_ledgers_by_hash(
-          "LedgerCache",
-          app_.config().getValueFor(SizedItem::ledgerSize),
-          std::chrono::seconds{app_.config().getValueFor(SizedItem::ledgerAge)},
-          stopwatch(),
-          app_.journal("TaggedCache"))
     , m_consensus_validated(
           "ConsensusValidated",
           64,
@@ -28,6 +20,13 @@ LedgerHistory::LedgerHistory(beast::insight::Collector::ptr const& collector, Ap
           app_.journal("TaggedCache"))
     , j_(app.journal("LedgerHistory"))
 {
+    auto lock = ledger_maps_.lock();
+    lock->by_hash = std::make_unique<LedgerMaps::LedgersByHash>(
+        "LedgerCache",
+        app_.config().getValueFor(SizedItem::ledgerSize),
+        std::chrono::seconds{app_.config().getValueFor(SizedItem::ledgerAge)},
+        stopwatch(),
+        app_.journal("TaggedCache"));
 }
 
 bool
@@ -39,12 +38,11 @@ LedgerHistory::insert(std::shared_ptr<Ledger const> const& ledger, bool validate
     XRPL_ASSERT(
         ledger->stateMap().getHash().isNonZero(), "xrpl::LedgerHistory::insert : nonzero hash");
 
-    std::unique_lock sl(m_ledgers_by_hash.peekMutex());
-
+    auto lock = ledger_maps_.lock();
     bool const alreadyHad =
-        m_ledgers_by_hash.canonicalize_replace_cache(ledger->header().hash, ledger);
+        lock->by_hash->canonicalize_replace_cache(ledger->header().hash, ledger);
     if (validated)
-        mLedgersByIndex[ledger->header().seq] = ledger->header().hash;
+        lock->by_index[ledger->header().seq] = ledger->header().hash;
 
     return alreadyHad;
 }
@@ -52,8 +50,8 @@ LedgerHistory::insert(std::shared_ptr<Ledger const> const& ledger, bool validate
 LedgerHash
 LedgerHistory::getLedgerHash(LedgerIndex index)
 {
-    std::unique_lock sl(m_ledgers_by_hash.peekMutex());
-    if (auto it = mLedgersByIndex.find(index); it != mLedgersByIndex.end())
+    auto lock = ledger_maps_.lock();
+    if (auto it = lock->by_index.find(index); it != lock->by_index.end())
         return it->second;
     return {};
 }
@@ -61,18 +59,22 @@ LedgerHistory::getLedgerHash(LedgerIndex index)
 std::shared_ptr<Ledger const>
 LedgerHistory::getLedgerBySeq(LedgerIndex index)
 {
+    uint256 hash;
     {
-        std::unique_lock sl(m_ledgers_by_hash.peekMutex());
-        auto it = mLedgersByIndex.find(index);
-
-        if (it != mLedgersByIndex.end())
+        auto lock = ledger_maps_.lock();
+        if (auto it = lock->by_index.find(index); it != lock->by_index.end())
         {
-            uint256 hash = it->second;
-            sl.unlock();
-            return getLedgerByHash(hash);
+            hash = it->second;
+        }
+        else
+        {
+            hash = {};
         }
     }
 
+    if (!hash.isZero())
+        return getLedgerByHash(hash);
+
     std::shared_ptr<Ledger const> ret = loadByIndex(index, app_);
 
     if (!ret)
@@ -83,12 +85,11 @@ LedgerHistory::getLedgerBySeq(LedgerIndex index)
 
     {
         // Add this ledger to the local tracking by index
-        std::unique_lock sl(m_ledgers_by_hash.peekMutex());
-
+        auto lock = ledger_maps_.lock();
         XRPL_ASSERT(
             ret->isImmutable(), "xrpl::LedgerHistory::getLedgerBySeq : immutable result ledger");
-        m_ledgers_by_hash.canonicalize_replace_client(ret->header().hash, ret);
-        mLedgersByIndex[ret->header().seq] = ret->header().hash;
+        lock->by_hash->canonicalize_replace_client(ret->header().hash, ret);
+        lock->by_index[ret->header().seq] = ret->header().hash;
         return (ret->header().seq == index) ? ret : nullptr;
     }
 }
@@ -96,7 +97,11 @@ LedgerHistory::getLedgerBySeq(LedgerIndex index)
 std::shared_ptr<Ledger const>
 LedgerHistory::getLedgerByHash(LedgerHash const& hash)
 {
-    auto ret = m_ledgers_by_hash.fetch(hash);
+    std::shared_ptr<Ledger const> ret;
+    {
+        auto lock = ledger_maps_.lock();
+        ret = lock->by_hash->fetch(hash);
+    }
 
     if (ret)
     {
@@ -121,7 +126,10 @@ LedgerHistory::getLedgerByHash(LedgerHash const& hash)
     XRPL_ASSERT(
         ret->header().hash == hash,
         "xrpl::LedgerHistory::getLedgerByHash : loaded ledger hash match");
-    m_ledgers_by_hash.canonicalize_replace_client(ret->header().hash, ret);
+    {
+        auto lock = ledger_maps_.lock();
+        lock->by_hash->canonicalize_replace_client(ret->header().hash, ret);
+    }
     XRPL_ASSERT(
         ret->header().hash == hash, "xrpl::LedgerHistory::getLedgerByHash : result hash match");
 
@@ -404,34 +412,31 @@ LedgerHistory::builtLedger(
     LedgerHash hash = ledger->header().hash;
     XRPL_ASSERT(!hash.isZero(), "xrpl::LedgerHistory::builtLedger : nonzero hash");
 
-    std::unique_lock sl(m_consensus_validated.peekMutex());
-
-    auto entry = std::make_shared<cv_entry>();
-    m_consensus_validated.canonicalize_replace_client(index, entry);
-
-    if (entry->validated && !entry->built)
-    {
-        if (entry->validated.value() != hash)
+    m_consensus_validated.fetch_and_modify(index, [&](cv_entry& entry) {
+        if (entry.validated && !entry.built)
         {
-            JLOG(j_.error()) << "MISMATCH: seq=" << index
-                             << " validated:" << entry->validated.value() << " then:" << hash;
-            handleMismatch(
-                hash,
-                entry->validated.value(),
-                consensusHash,
-                entry->validatedConsensusHash,
-                consensus);
+            if (entry.validated.value() != hash)
+            {
+                JLOG(j_.error()) << "MISMATCH: seq=" << index
+                                 << " validated:" << entry.validated.value() << " then:" << hash;
+                handleMismatch(
+                    hash,
+                    entry.validated.value(),
+                    consensusHash,
+                    entry.validatedConsensusHash,
+                    consensus);
+            }
+            else
+            {
+                // We validated a ledger and then built it locally
+                JLOG(j_.debug()) << "MATCH: seq=" << index << " late";
+            }
         }
-        else
-        {
-            // We validated a ledger and then built it locally
-            JLOG(j_.debug()) << "MATCH: seq=" << index << " late";
-        }
-    }
 
-    entry->built.emplace(hash);
-    entry->builtConsensusHash.emplace(consensusHash);
-    entry->consensus.emplace(std::move(consensus));
+        entry.built.emplace(hash);
+        entry.builtConsensusHash.emplace(consensusHash);
+        entry.consensus.emplace(std::move(consensus));
+    });
 }
 
 void
@@ -443,33 +448,30 @@ LedgerHistory::validatedLedger(
     LedgerHash hash = ledger->header().hash;
     XRPL_ASSERT(!hash.isZero(), "xrpl::LedgerHistory::validatedLedger : nonzero hash");
 
-    std::unique_lock sl(m_consensus_validated.peekMutex());
-
-    auto entry = std::make_shared<cv_entry>();
-    m_consensus_validated.canonicalize_replace_client(index, entry);
-
-    if (entry->built && !entry->validated)
-    {
-        if (entry->built.value() != hash)
+    m_consensus_validated.fetch_and_modify(index, [&](cv_entry& entry) {
+        if (entry.built && !entry.validated)
         {
-            JLOG(j_.error()) << "MISMATCH: seq=" << index << " built:" << entry->built.value()
-                             << " then:" << hash;
-            handleMismatch(
-                entry->built.value(),
-                hash,
-                entry->builtConsensusHash,
-                consensusHash,
-                entry->consensus.value());
+            if (entry.built.value() != hash)
+            {
+                JLOG(j_.error()) << "MISMATCH: seq=" << index << " built:" << entry.built.value()
+                                 << " then:" << hash;
+                handleMismatch(
+                    entry.built.value(),
+                    hash,
+                    entry.builtConsensusHash,
+                    consensusHash,
+                    entry.consensus.value());
+            }
+            else
+            {
+                // We built a ledger locally and then validated it
+                JLOG(j_.debug()) << "MATCH: seq=" << index;
+            }
         }
-        else
-        {
-            // We built a ledger locally and then validated it
-            JLOG(j_.debug()) << "MATCH: seq=" << index;
-        }
-    }
 
-    entry->validated.emplace(hash);
-    entry->validatedConsensusHash = consensusHash;
+        entry.validated.emplace(hash);
+        entry.validatedConsensusHash = consensusHash;
+    });
 }
 
 /** Ensure m_ledgers_by_hash doesn't have the wrong hash for a particular index
@@ -477,13 +479,14 @@ LedgerHistory::validatedLedger(
 bool
 LedgerHistory::fixIndex(LedgerIndex ledgerIndex, LedgerHash const& ledgerHash)
 {
-    std::unique_lock sl(m_ledgers_by_hash.peekMutex());
-    auto it = mLedgersByIndex.find(ledgerIndex);
-
-    if ((it != mLedgersByIndex.end()) && (it->second != ledgerHash))
+    auto lock = ledger_maps_.lock();
+    if (auto it = lock->by_index.find(ledgerIndex); it != lock->by_index.end())
     {
-        it->second = ledgerHash;
-        return false;
+        if (it->second != ledgerHash)
+        {
+            lock->by_index[ledgerIndex] = ledgerHash;
+            return false;
+        }
     }
     return true;
 }
@@ -491,12 +494,45 @@ LedgerHistory::fixIndex(LedgerIndex ledgerIndex, LedgerHash const& ledgerHash)
 void
 LedgerHistory::clearLedgerCachePrior(LedgerIndex seq)
 {
-    for (LedgerHash it : m_ledgers_by_hash.getKeys())
+    std::size_t hashesCleared = 0;
+    std::size_t indexesCleared = 0;
+    std::size_t cacheSize = 0;
+    std::size_t indexSize = 0;
+
     {
-        auto const ledger = getLedgerByHash(it);
-        if (!ledger || ledger->header().seq < seq)
-            m_ledgers_by_hash.del(it, false);
+        auto lock = ledger_maps_.lock();
+        for (LedgerHash it : lock->by_hash->getKeys())
+        {
+            auto const ledger = getLedgerByHash(it);
+            if (!ledger || ledger->header().seq < seq)
+            {
+                lock->by_hash->del(it, false);
+                ++hashesCleared;
+            }
+        }
+        cacheSize = lock->by_hash->size();
+
+        auto it = lock->by_index.begin();
+        while (it != lock->by_index.end())
+        {
+            if (it->first < seq)
+            {
+                it = lock->by_index.erase(it);
+                ++indexesCleared;
+            }
+            else
+            {
+                ++it;
+            }
+        }
+        indexSize = lock->by_index.size();
     }
+
+    JLOG(j_.debug()) << "LedgersByHash: cleared " << hashesCleared << " entries before seq " << seq
+                     << " (total now " << cacheSize << ")";
+
+    JLOG(j_.debug()) << "LedgersByIndex: cleared " << indexesCleared << " index entries before seq "
+                     << seq << " (total now " << indexSize << ")";
 }
 
 }  // namespace xrpl

src/xrpld/app/ledger/LedgerHistory.h

@@ -3,9 +3,12 @@
 #include <xrpld/app/ledger/Ledger.h>
 #include <xrpld/app/main/Application.h>
 
+#include <xrpl/basics/Mutex.hpp>
 #include <xrpl/beast/insight/Collector.h>
 #include <xrpl/protocol/RippleLedgerHash.h>
 
+#include <map>
+#include <memory>
 #include <optional>
 
 namespace xrpl {
@@ -30,7 +33,8 @@ public:
     float
     getCacheHitRate()
     {
-        return m_ledgers_by_hash.getHitRate();
+        auto lock = ledger_maps_.lock();
+        return lock->by_hash->getHitRate();
     }
 
     /** Get a ledger given its sequence number */
@@ -53,7 +57,10 @@ public:
     void
     sweep()
     {
-        m_ledgers_by_hash.sweep();
+        {
+            auto lock = ledger_maps_.lock();
+            lock->by_hash->sweep();
+        }
         m_consensus_validated.sweep();
     }
 
@@ -101,9 +108,15 @@ private:
     beast::insight::Collector::ptr collector_;
     beast::insight::Counter mismatch_counter_;
 
-    using LedgersByHash = TaggedCache<LedgerHash, Ledger const>;
+    struct LedgerMaps
+    {
+        using LedgersByHash = TaggedCache<LedgerHash, Ledger const>;
 
-    LedgersByHash m_ledgers_by_hash;
+        std::unique_ptr<LedgersByHash> by_hash;
+        std::map<LedgerIndex, LedgerHash> by_index;  // validated ledgers
+    };
+
+    xrpl::Mutex<LedgerMaps, std::recursive_mutex> ledger_maps_;
 
     // Maps ledger indexes to the corresponding hashes
     // For debug and logging purposes
@@ -123,9 +136,6 @@ private:
     using ConsensusValidated = TaggedCache<LedgerIndex, cv_entry>;
     ConsensusValidated m_consensus_validated;
 
-    // Maps ledger indexes to the corresponding hash.
-    std::map<LedgerIndex, LedgerHash> mLedgersByIndex;  // validated ledgers
-
     beast::Journal j_;
 };