fix: bound history priming ledger residency

- Search both LedgerMaster and InboundLedgers for the closest fully wired base.
- Implement sameChainDistance helper to accurately calculate distance between ledgers on the same chain.
- Use findBestFullyWiredBase to minimize the 'prime walk' delta.

include/xrpl/basics/ReaderPreferringSharedMutex.h

@@ -0,0 +1,106 @@
+#pragma once
+
+#include <shared_mutex>
+
+// On Linux (glibc), std::shared_mutex wraps pthread_rwlock_t initialised
+// with PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP.  This means a
+// pending exclusive lock() blocks new shared (reader) acquisitions,
+// causing reader starvation when writers contend frequently.
+//
+// On macOS / ARM (libc++), std::shared_mutex is already reader-preferring,
+// so the same code behaves differently across platforms.
+//
+// This header provides reader_preferring_shared_mutex:
+//   - On Linux it wraps pthread_rwlock_t initialised with
+//     PTHREAD_RWLOCK_PREFER_READER_NP, matching macOS semantics.
+//   - On all other platforms it is a type alias for std::shared_mutex.
+//
+// The interface is identical to std::shared_mutex, so it works with
+// std::shared_lock and std::unique_lock.
+
+#if defined(__linux__)
+
+#include <cerrno>
+#include <pthread.h>
+#include <stdexcept>
+
+namespace ripple {
+
+class reader_preferring_shared_mutex
+{
+    pthread_rwlock_t rwlock_;
+
+public:
+    reader_preferring_shared_mutex()
+    {
+        pthread_rwlockattr_t attr;
+        pthread_rwlockattr_init(&attr);
+        pthread_rwlockattr_setkind_np(&attr, PTHREAD_RWLOCK_PREFER_READER_NP);
+        int rc = pthread_rwlock_init(&rwlock_, &attr);
+        pthread_rwlockattr_destroy(&attr);
+        if (rc != 0)
+            throw std::system_error(
+                rc, std::system_category(), "pthread_rwlock_init");
+    }
+
+    ~reader_preferring_shared_mutex()
+    {
+        pthread_rwlock_destroy(&rwlock_);
+    }
+
+    reader_preferring_shared_mutex(reader_preferring_shared_mutex const&) =
+        delete;
+    reader_preferring_shared_mutex&
+    operator=(reader_preferring_shared_mutex const&) = delete;
+
+    // Exclusive (writer) locking
+    void
+    lock()
+    {
+        pthread_rwlock_wrlock(&rwlock_);
+    }
+
+    bool
+    try_lock()
+    {
+        return pthread_rwlock_trywrlock(&rwlock_) == 0;
+    }
+
+    void
+    unlock()
+    {
+        pthread_rwlock_unlock(&rwlock_);
+    }
+
+    // Shared (reader) locking
+    void
+    lock_shared()
+    {
+        pthread_rwlock_rdlock(&rwlock_);
+    }
+
+    bool
+    try_lock_shared()
+    {
+        return pthread_rwlock_tryrdlock(&rwlock_) == 0;
+    }
+
+    void
+    unlock_shared()
+    {
+        pthread_rwlock_unlock(&rwlock_);
+    }
+};
+
+}  // namespace ripple
+
+#else  // !__linux__
+
+namespace ripple {
+
+// macOS, Windows, etc. — std::shared_mutex is already reader-preferring.
+using reader_preferring_shared_mutex = std::shared_mutex;
+
+}  // namespace ripple
+
+#endif

src/libxrpl/basics/contract.cpp

@@ -20,8 +20,13 @@
 #include <xrpl/basics/Log.h>
 #include <xrpl/basics/contract.h>
 #include <xrpl/beast/utility/instrumentation.h>
+#ifndef BOOST_STACKTRACE_GNU_SOURCE_NOT_REQUIRED
+#define BOOST_STACKTRACE_GNU_SOURCE_NOT_REQUIRED
+#endif
+#include <boost/stacktrace.hpp>
 #include <cstdlib>
 #include <iostream>
+#include <sstream>
 
 namespace ripple {
 
@@ -41,7 +46,12 @@ accessViolation() noexcept
 void
 LogThrow(std::string const& title)
 {
-    JLOG(debugLog().warn()) << title;
+    std::ostringstream oss;
+    oss << title << '\n' << boost::stacktrace::stacktrace();
+    JLOG(debugLog().warn()) << oss.str();
+    // Also mirror to stderr so uncaught exceptions leave a trace even when
+    // log output is buffered/lost before terminate().
+    std::cerr << oss.str() << std::endl;
 }
 
 [[noreturn]] void

src/xrpld/app/ledger/InboundLedgers.h

@@ -83,9 +83,13 @@ public:
     virtual std::size_t
     fetchRate() = 0;
 
-    /** Called when a complete ledger is obtained. */
+    /** Called when a complete history ledger is obtained. */
     virtual void
-    onLedgerFetched() = 0;
+    onLedgerFetched(std::shared_ptr<InboundLedger> const& inbound) = 0;
+
+    virtual std::shared_ptr<Ledger const>
+    getClosestFullyWiredLedger(
+        std::shared_ptr<Ledger const> const& targetLedger) = 0;
 
     virtual void
     gotFetchPack() = 0;

src/xrpld/app/ledger/Ledger.cpp

@@ -50,6 +50,8 @@
 #include <xrpl/protocol/digest.h>
 #include <xrpl/protocol/jss.h>
 #include <boost/optional.hpp>
+#include <cstdlib>
+#include <string_view>
 #include <utility>
 #include <vector>
 
@@ -59,6 +61,33 @@ namespace ripple {
 
 create_genesis_t const create_genesis{};
 
+namespace {
+
+bool
+isRWDBNullMode()
+{
+    static bool const v = [] {
+        char const* e = std::getenv("XAHAU_RWDB_NULL");
+        return e && *e && std::string_view{e} != "0";
+    }();
+    return v;
+}
+
+template <class Map>
+std::size_t
+wireCompleteSHAMap(Map const& map)
+{
+    std::size_t leaves = 0;
+    for (auto const& item : map)
+    {
+        (void)item;
+        ++leaves;
+    }
+    return leaves;
+}
+
+}  // namespace
+
 uint256
 calculateLedgerHash(LedgerInfo const& info)
 {
@@ -249,6 +278,7 @@ Ledger::Ledger(
 
     stateMap_.flushDirty(hotACCOUNT_NODE);
     setImmutable();
+    setFullyWired();
 }
 
 Ledger::Ledger(
@@ -313,6 +343,7 @@ Ledger::Ledger(
 // Create a new ledger that follows this one
 Ledger::Ledger(Ledger const& prevLedger, NetClock::time_point closeTime)
     : mImmutable(false)
+    , fullyWired_(prevLedger.isFullyWired())
     , txMap_(SHAMapType::TRANSACTION, prevLedger.txMap_.family())
     , stateMap_(prevLedger.stateMap_, true)
     , fees_(prevLedger.fees_)
@@ -390,6 +421,30 @@ Ledger::setImmutable(bool rehash)
     setup();
 }
 
+bool
+Ledger::fullWireForUse(beast::Journal journal, char const* context) const
+{
+    if (!isRWDBNullMode() || isFullyWired())
+        return true;
+
+    try
+    {
+        auto const stateLeaves = wireCompleteSHAMap(stateMap_);
+        auto const txLeaves = wireCompleteSHAMap(txMap_);
+        setFullyWired();
+        JLOG(journal.info())
+            << context << ": fully wired ledger " << info_.seq << " ("
+            << stateLeaves << " state leaves, " << txLeaves << " tx leaves)";
+        return true;
+    }
+    catch (SHAMapMissingNode const& e)
+    {
+        JLOG(journal.warn()) << context << ": incomplete ledger " << info_.seq
+                             << ": " << e.what();
+        return false;
+    }
+}
+
 // raw setters for catalogue
 void
 Ledger::setCloseFlags(int closeFlags)
@@ -1130,14 +1185,17 @@ loadLedgerHelper(LedgerInfo const& info, Application& app, bool acquire)
 }
 
 static void
-finishLoadByIndexOrHash(
-    std::shared_ptr<Ledger> const& ledger,
-    Config const& config,
-    beast::Journal j)
+finishLoadByIndexOrHash(std::shared_ptr<Ledger>& ledger, beast::Journal j)
 {
     if (!ledger)
         return;
 
+    if (!ledger->fullWireForUse(j, "finishLoadByIndexOrHash"))
+    {
+        ledger.reset();
+        return;
+    }
+
     XRPL_ASSERT(
         ledger->read(keylet::fees()),
         "ripple::finishLoadByIndexOrHash : valid ledger fees");
@@ -1155,7 +1213,13 @@ getLatestLedger(Application& app)
         app.getRelationalDatabase().getNewestLedgerInfo();
     if (!info)
         return {std::shared_ptr<Ledger>(), {}, {}};
-    return {loadLedgerHelper(*info, app, true), info->seq, info->hash};
+    auto ledger = loadLedgerHelper(*info, app, true);
+    if (ledger &&
+        !ledger->fullWireForUse(app.journal("Ledger"), "getLatestLedger"))
+    {
+        ledger.reset();
+    }
+    return {ledger, info->seq, info->hash};
 }
 
 std::shared_ptr<Ledger>
@@ -1165,7 +1229,7 @@ loadByIndex(std::uint32_t ledgerIndex, Application& app, bool acquire)
             app.getRelationalDatabase().getLedgerInfoByIndex(ledgerIndex))
     {
         std::shared_ptr<Ledger> ledger = loadLedgerHelper(*info, app, acquire);
-        finishLoadByIndexOrHash(ledger, app.config(), app.journal("Ledger"));
+        finishLoadByIndexOrHash(ledger, app.journal("Ledger"));
         return ledger;
     }
     return {};
@@ -1178,7 +1242,7 @@ loadByHash(uint256 const& ledgerHash, Application& app, bool acquire)
             app.getRelationalDatabase().getLedgerInfoByHash(ledgerHash))
     {
         std::shared_ptr<Ledger> ledger = loadLedgerHelper(*info, app, acquire);
-        finishLoadByIndexOrHash(ledger, app.config(), app.journal("Ledger"));
+        finishLoadByIndexOrHash(ledger, app.journal("Ledger"));
         XRPL_ASSERT(
             !ledger || ledger->info().hash == ledgerHash,
             "ripple::loadByHash : ledger hash match if loaded");

src/xrpld/app/ledger/Ledger.h

@@ -31,6 +31,7 @@
 #include <xrpl/protocol/STLedgerEntry.h>
 #include <xrpl/protocol/Serializer.h>
 #include <xrpl/protocol/TxMeta.h>
+#include <atomic>
 #include <mutex>
 
 namespace ripple {
@@ -294,6 +295,21 @@ public:
         return mImmutable;
     }
 
+    bool
+    isFullyWired() const
+    {
+        return fullyWired_.load(std::memory_order_acquire);
+    }
+
+    void
+    setFullyWired() const
+    {
+        fullyWired_.store(true, std::memory_order_release);
+    }
+
+    bool
+    fullWireForUse(beast::Journal journal, char const* context) const;
+
     /*  Mark this ledger as "should be full".
 
         "Full" is metadata property of the ledger, it indicates
@@ -417,6 +433,7 @@ private:
     defaultFees(Config const& config);
 
     bool mImmutable;
+    mutable std::atomic<bool> fullyWired_{false};
 
     // A SHAMap containing the transactions associated with this ledger.
     SHAMap mutable txMap_;

src/xrpld/app/ledger/LedgerMaster.h

@@ -37,6 +37,7 @@
 #include <xrpl/protocol/RippleLedgerHash.h>
 #include <xrpl/protocol/STValidation.h>
 #include <xrpl/protocol/messages.h>
+#include <deque>
 #include <optional>
 
 #include <mutex>
@@ -183,6 +184,10 @@ public:
     std::shared_ptr<Ledger const>
     getLedgerByHash(uint256 const& hash);
 
+    std::shared_ptr<Ledger const>
+    getClosestFullyWiredLedger(
+        std::shared_ptr<Ledger const> const& targetLedger);
+
     void
     setLedgerRangePresent(
         std::uint32_t minV,
@@ -347,6 +352,12 @@ private:
     // The last ledger we handled fetching history
     std::shared_ptr<Ledger const> mHistLedger;
 
+    // Sliding window of recently validated ledgers pinned in memory so their
+    // SHAMap state trees remain reachable via shared_ptr. Required when the
+    // node store does not persist state nodes (e.g. RWDB with
+    // XAHAU_RWDB_DISCARD_HOT_ACCOUNT_NODE). Guarded by m_mutex.
+    std::deque<std::shared_ptr<Ledger const>> mRetainedLedgers;
+
     // Fully validated ledger, whether or not we have the ledger resident.
     std::pair<uint256, LedgerIndex> mLastValidLedger{uint256(), 0};
 

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

@@ -35,12 +35,169 @@
 #include <boost/iterator/function_output_iterator.hpp>
 
 #include <algorithm>
+#include <cstdlib>
+#include <limits>
 #include <random>
+#include <string_view>
 
 namespace ripple {
 
 using namespace std::chrono_literals;
 
+namespace {
+
+bool
+isRWDBNullMode()
+{
+    static bool const v = [] {
+        char const* e = std::getenv("XAHAU_RWDB_NULL");
+        return e && *e && std::string_view{e} != "0";
+    }();
+    return v;
+}
+
+template <class Map>
+std::size_t
+wireCompleteSHAMap(Map const& map)
+{
+    std::size_t leaves = 0;
+    for (auto const& item : map)
+    {
+        (void)item;
+        ++leaves;
+    }
+    return leaves;
+}
+
+std::optional<std::uint32_t>
+sameChainDistance(
+    std::shared_ptr<Ledger const> const& targetLedger,
+    std::shared_ptr<Ledger const> const& candidate,
+    beast::Journal journal)
+{
+    if (!targetLedger || !candidate || !candidate->isFullyWired())
+        return std::nullopt;
+
+    if (candidate->info().hash == targetLedger->info().hash)
+        return std::nullopt;
+
+    bool sameChain = false;
+    try
+    {
+        if (candidate->info().seq < targetLedger->info().seq)
+        {
+            if (auto const hash =
+                    hashOfSeq(*targetLedger, candidate->info().seq, journal);
+                hash && *hash == candidate->info().hash)
+            {
+                sameChain = true;
+            }
+        }
+        else if (candidate->info().seq > targetLedger->info().seq)
+        {
+            if (auto const hash =
+                    hashOfSeq(*candidate, targetLedger->info().seq, journal);
+                hash && *hash == targetLedger->info().hash)
+            {
+                sameChain = true;
+            }
+        }
+    }
+    catch (std::exception const&)
+    {
+        sameChain = false;
+    }
+
+    if (!sameChain)
+        return std::nullopt;
+
+    return candidate->info().seq < targetLedger->info().seq
+        ? targetLedger->info().seq - candidate->info().seq
+        : candidate->info().seq - targetLedger->info().seq;
+}
+
+std::shared_ptr<Ledger const>
+chooseCloserBase(
+    std::shared_ptr<Ledger const> const& targetLedger,
+    std::shared_ptr<Ledger const> const& first,
+    std::shared_ptr<Ledger const> const& second,
+    beast::Journal journal)
+{
+    auto const firstDistance = sameChainDistance(targetLedger, first, journal);
+    auto const secondDistance =
+        sameChainDistance(targetLedger, second, journal);
+
+    if (firstDistance && secondDistance)
+        return *firstDistance <= *secondDistance ? first : second;
+    if (firstDistance)
+        return first;
+    if (secondDistance)
+        return second;
+    return {};
+}
+
+std::shared_ptr<Ledger const>
+findBestFullyWiredBase(
+    Application& app,
+    std::shared_ptr<Ledger const> const& targetLedger,
+    beast::Journal journal)
+{
+    auto const ledgerMasterBase =
+        app.getLedgerMaster().getClosestFullyWiredLedger(targetLedger);
+    auto const inboundBase =
+        app.getInboundLedgers().getClosestFullyWiredLedger(targetLedger);
+    return chooseCloserBase(
+        targetLedger, inboundBase, ledgerMasterBase, journal);
+}
+
+bool
+primeInboundLedgerForUse(
+    std::shared_ptr<Ledger> const& ledger,
+    std::shared_ptr<Ledger const> const& baseLedger,
+    beast::Journal journal,
+    char const* context)
+{
+    if (!isRWDBNullMode())
+        return true;
+
+    if (ledger->isFullyWired())
+        return true;
+
+    if (!baseLedger || !baseLedger->isFullyWired())
+    {
+        return ledger->fullWireForUse(journal, context);
+    }
+
+    try
+    {
+        std::size_t stateNodes = 0;
+        // By the time an inbound ledger is marked complete, sync has already
+        // descended the current tree; this delta walk avoids rewalking
+        // unchanged state subtrees that are known-good via a fully wired
+        // same-chain base ledger.
+        ledger->stateMap().visitDifferences(
+            &baseLedger->stateMap(), [&stateNodes](SHAMapTreeNode const&) {
+                ++stateNodes;
+                return true;
+            });
+        auto const txLeaves = wireCompleteSHAMap(ledger->txMap());
+        ledger->setFullyWired();
+        JLOG(journal.info())
+            << context << ": fully wired ledger " << ledger->info().seq << " ("
+            << stateNodes << " changed state nodes vs base ledger "
+            << baseLedger->info().seq << ", " << txLeaves << " tx leaves)";
+        return true;
+    }
+    catch (SHAMapMissingNode const& e)
+    {
+        JLOG(journal.warn()) << context << ": incomplete ledger "
+                             << ledger->info().seq << ": " << e.what();
+        return false;
+    }
+}
+
+}  // namespace
+
 enum {
     // Number of peers to start with
     peerCountStart = 5
@@ -120,13 +277,25 @@ InboundLedger::init(ScopedLockType& collectionLock)
 
     JLOG(journal_.debug()) << "Acquiring ledger we already have in "
                            << " local store. " << hash_;
+    auto const baseLedger = findBestFullyWiredBase(app_, mLedger, journal_);
+    if (!primeInboundLedgerForUse(
+            mLedger, baseLedger, journal_, "InboundLedger::init"))
+    {
+        complete_ = false;
+        failed_ = true;
+        done();
+        return;
+    }
     XRPL_ASSERT(
         mLedger->read(keylet::fees()),
         "ripple::InboundLedger::init : valid ledger fees");
     mLedger->setImmutable();
 
     if (mReason == Reason::HISTORY)
+    {
+        app_.getInboundLedgers().onLedgerFetched(shared_from_this());
         return;
+    }
 
     app_.getLedgerMaster().storeLedger(mLedger);
 
@@ -351,10 +520,6 @@ InboundLedger::tryDB(NodeStore::Database& srcDB)
     {
         JLOG(journal_.debug()) << "Had everything locally";
         complete_ = true;
-        XRPL_ASSERT(
-            mLedger->read(keylet::fees()),
-            "ripple::InboundLedger::tryDB : valid ledger fees");
-        mLedger->setImmutable();
     }
 }
 
@@ -453,18 +618,29 @@ InboundLedger::done()
 
     if (complete_ && !failed_ && mLedger)
     {
-        XRPL_ASSERT(
-            mLedger->read(keylet::fees()),
-            "ripple::InboundLedger::done : valid ledger fees");
-        mLedger->setImmutable();
-        switch (mReason)
+        auto const baseLedger = findBestFullyWiredBase(app_, mLedger, journal_);
+        if (!primeInboundLedgerForUse(
+                mLedger, baseLedger, journal_, "InboundLedger::done"))
         {
-            case Reason::HISTORY:
-                app_.getInboundLedgers().onLedgerFetched();
-                break;
-            default:
-                app_.getLedgerMaster().storeLedger(mLedger);
-                break;
+            complete_ = false;
+            failed_ = true;
+        }
+        else
+        {
+            XRPL_ASSERT(
+                mLedger->read(keylet::fees()),
+                "ripple::InboundLedger::done : valid ledger fees");
+            mLedger->setImmutable();
+
+            switch (mReason)
+            {
+                case Reason::HISTORY:
+                    app_.getInboundLedgers().onLedgerFetched(shared_from_this());
+                    break;
+                default:
+                    app_.getLedgerMaster().storeLedger(mLedger);
+                    break;
+            }
         }
     }
 
@@ -473,6 +649,42 @@ InboundLedger::done()
         jtLEDGER_DATA, "AcquisitionDone", [self = shared_from_this()]() {
             if (self->complete_ && !self->failed_)
             {
+                if (!isRWDBNullMode() && self->mReason != Reason::HISTORY)
+                {
+                    // Prime the state tree BEFORE checkAccept so consensus
+                    // never sees a lazy tree. Runs off any inbound lock —
+                    // this job is dispatched without mtx_ held.
+                    // visitDifferences against prior validated walks only
+                    // the delta; canonicalization means shared subtrees are
+                    // the same inner objects (already wired). Gated on
+                    // non-HISTORY to avoid paying on historical backfills.
+                    auto const prior =
+                        self->app_.getLedgerMaster().getValidatedLedger();
+                    SHAMap const* have = prior ? &prior->stateMap() : nullptr;
+
+                    try
+                    {
+                        std::size_t walked = 0;
+                        self->mLedger->stateMap().visitDifferences(
+                            have, [&walked](SHAMapTreeNode const&) {
+                                ++walked;
+                                return true;
+                            });
+                        JLOG(self->journal_.info())
+                            << "Inbound prime: ledger "
+                            << self->mLedger->info().seq << " wired " << walked
+                            << (have ? " delta nodes vs prior validated"
+                                     : " nodes (first full walk)");
+                    }
+                    catch (SHAMapMissingNode const& e)
+                    {
+                        JLOG(self->journal_.warn())
+                            << "Inbound prime: incomplete state tree for "
+                            << "ledger " << self->mLedger->info().seq << ": "
+                            << e.what();
+                    }
+                }
+
                 self->app_.getLedgerMaster().checkAccept(self->getLedger());
                 self->app_.getLedgerMaster().tryAdvance();
             }

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

@@ -22,6 +22,7 @@
 #include <xrpld/app/main/Application.h>
 #include <xrpld/app/misc/NetworkOPs.h>
 #include <xrpld/core/JobQueue.h>
+#include <xrpld/ledger/View.h>
 #include <xrpld/perflog/PerfLog.h>
 #include <xrpl/basics/DecayingSample.h>
 #include <xrpl/basics/Log.h>
@@ -30,13 +31,75 @@
 #include <xrpl/beast/core/LexicalCast.h>
 #include <xrpl/protocol/jss.h>
 
+#include <deque>
 #include <exception>
+#include <limits>
 #include <memory>
 #include <mutex>
 #include <vector>
 
 namespace ripple {
 
+namespace {
+
+std::size_t
+historyPrimingCacheSize(Application& app)
+{
+    auto const configured = static_cast<std::size_t>(app.config().LEDGER_HISTORY);
+    auto const bounded = std::min<std::size_t>(
+        configured == 0 ? 8 : configured, 32);
+    return std::max<std::size_t>(1, bounded);
+}
+
+std::optional<std::uint32_t>
+sameChainDistance(
+    std::shared_ptr<Ledger const> const& targetLedger,
+    std::shared_ptr<Ledger const> const& candidate,
+    beast::Journal journal)
+{
+    if (!targetLedger || !candidate || !candidate->isFullyWired())
+        return std::nullopt;
+
+    if (candidate->info().hash == targetLedger->info().hash)
+        return std::nullopt;
+
+    bool sameChain = false;
+    try
+    {
+        if (candidate->info().seq < targetLedger->info().seq)
+        {
+            if (auto const hash =
+                    hashOfSeq(*targetLedger, candidate->info().seq, journal);
+                hash && *hash == candidate->info().hash)
+            {
+                sameChain = true;
+            }
+        }
+        else if (candidate->info().seq > targetLedger->info().seq)
+        {
+            if (auto const hash =
+                    hashOfSeq(*candidate, targetLedger->info().seq, journal);
+                hash && *hash == targetLedger->info().hash)
+            {
+                sameChain = true;
+            }
+        }
+    }
+    catch (std::exception const&)
+    {
+        sameChain = false;
+    }
+
+    if (!sameChain)
+        return std::nullopt;
+
+    return candidate->info().seq < targetLedger->info().seq
+        ? targetLedger->info().seq - candidate->info().seq
+        : candidate->info().seq - targetLedger->info().seq;
+}
+
+}  // namespace
+
 class InboundLedgersImp : public InboundLedgers
 {
 private:
@@ -61,6 +124,7 @@ public:
         , m_clock(clock)
         , mRecentFailures(clock)
         , mCounter(collector->make_counter("ledger_fetches"))
+        , historyPrimingCacheSize_(historyPrimingCacheSize(app))
         , mPeerSetBuilder(std::move(peerSetBuilder))
     {
     }
@@ -296,6 +360,7 @@ public:
         ScopedLockType sl(mLock);
 
         mRecentFailures.clear();
+        recentHistoryLedgers_.clear();
         mLedgers.clear();
     }
 
@@ -306,15 +371,80 @@ public:
         return 60 * fetchRate_.value(m_clock.now());
     }
 
-    // Should only be called with an inboundledger that has
-    // a reason of history
+    // Should only be called with a complete inbound ledger that has
+    // a reason of history.
     void
-    onLedgerFetched() override
+    onLedgerFetched(std::shared_ptr<InboundLedger> const& inbound) override
     {
+        if (!inbound)
+            return;
+
+        auto const ledger = inbound->getLedger();
+        if (!ledger || !ledger->isFullyWired())
+            return;
+
+        {
+            ScopedLockType sl(mLock);
+            if (auto const it = mLedgers.find(ledger->info().hash);
+                it != mLedgers.end() && it->second.get() == inbound.get())
+            {
+                mLedgers.erase(it);
+            }
+
+            for (auto it = recentHistoryLedgers_.begin();
+                 it != recentHistoryLedgers_.end();)
+            {
+                if (!*it || (*it)->info().hash == ledger->info().hash)
+                    it = recentHistoryLedgers_.erase(it);
+                else
+                    ++it;
+            }
+
+            recentHistoryLedgers_.push_back(ledger);
+            while (recentHistoryLedgers_.size() > historyPrimingCacheSize_)
+                recentHistoryLedgers_.pop_front();
+        }
+
         std::lock_guard lock(fetchRateMutex_);
         fetchRate_.add(1, m_clock.now());
     }
 
+    std::shared_ptr<Ledger const>
+    getClosestFullyWiredLedger(
+        std::shared_ptr<Ledger const> const& targetLedger) override
+    {
+        if (!targetLedger)
+            return {};
+
+        std::vector<std::shared_ptr<Ledger const>> candidates;
+        {
+            ScopedLockType sl(mLock);
+            candidates.reserve(recentHistoryLedgers_.size());
+            for (auto const& ledger : recentHistoryLedgers_)
+            {
+                if (ledger && ledger->isFullyWired())
+                {
+                    candidates.push_back(ledger);
+                }
+            }
+        }
+
+        std::shared_ptr<Ledger const> best;
+        auto bestDistance = std::numeric_limits<std::uint32_t>::max();
+        for (auto const& candidate : candidates)
+        {
+            if (auto const distance =
+                    sameChainDistance(targetLedger, candidate, j_);
+                distance && *distance < bestDistance)
+            {
+                best = candidate;
+                bestDistance = *distance;
+            }
+        }
+
+        return best;
+    }
+
     Json::Value
     getInfo() override
     {
@@ -434,6 +564,7 @@ public:
     {
         ScopedLockType lock(mLock);
         stopping_ = true;
+        recentHistoryLedgers_.clear();
         mLedgers.clear();
         mRecentFailures.clear();
     }
@@ -454,10 +585,12 @@ private:
     bool stopping_ = false;
     using MapType = hash_map<uint256, std::shared_ptr<InboundLedger>>;
     MapType mLedgers;
+    std::deque<std::shared_ptr<Ledger const>> recentHistoryLedgers_;
 
     beast::aged_map<uint256, std::uint32_t> mRecentFailures;
 
     beast::insight::Counter mCounter;
+    std::size_t const historyPrimingCacheSize_;
 
     std::unique_ptr<PeerSetBuilder> mPeerSetBuilder;
 

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

@@ -697,11 +697,12 @@ LedgerMaster::tryFill(std::shared_ptr<Ledger const> ledger)
             if (it == ledgerHashes.end())
                 break;
 
+            auto const& firstHash = ledgerHashes.begin()->second.ledgerHash;
             if (!nodeStore.fetchNodeObject(
-                    ledgerHashes.begin()->second.ledgerHash,
-                    ledgerHashes.begin()->first))
+                    firstHash, ledgerHashes.begin()->first) &&
+                !getLedgerByHash(firstHash))
             {
-                // The ledger is not backed by the node store
+                // Not in node store and not in memory — genuinely missing
                 JLOG(m_journal.warn()) << "SQL DB ledger sequence " << seq
                                        << " mismatches node store";
                 break;
@@ -865,6 +866,44 @@ LedgerMaster::setFullLedger(
         mCompleteLedgers.insert(ledger->info().seq);
     }
 
+    // Pin a sliding window of recently validated current ledgers so their
+    // SHAMap state trees stay resident via shared_ptr. This tracks the
+    // server's active online band rather than retaining arbitrary historical
+    // backfill ledgers.
+    if (isCurrent && ledger_history_ > 0)
+    {
+        std::lock_guard ml(m_mutex);
+        bool const isFirst = mRetainedLedgers.empty();
+        mRetainedLedgers.push_back(ledger);
+        while (mRetainedLedgers.size() > ledger_history_)
+            mRetainedLedgers.pop_front();
+
+        // Legacy bootstrap for lazy trees. In null mode the ledger has
+        // already been fully wired before it reaches retention, so there is
+        // nothing left to do here.
+        if (isFirst && !ledger->isFullyWired())
+        {
+            try
+            {
+                std::size_t leafCount = 0;
+                for (auto const& item : ledger->stateMap())
+                {
+                    (void)item;
+                    ++leafCount;
+                }
+                JLOG(m_journal.info())
+                    << "Retention: primed state tree for ledger "
+                    << ledger->info().seq << " (" << leafCount << " leaves)";
+            }
+            catch (SHAMapMissingNode const& e)
+            {
+                JLOG(m_journal.warn())
+                    << "Retention: incomplete state tree for ledger "
+                    << ledger->info().seq << ": " << e.what();
+            }
+        }
+    }
+
     {
         std::lock_guard ml(m_mutex);
 
@@ -1663,6 +1702,12 @@ LedgerMaster::getCloseTimeByHash(
     LedgerHash const& ledgerHash,
     std::uint32_t index)
 {
+    // Prefer an in-memory Ledger (retained / history cache) over the node
+    // store so this works in RWDB-only configs where headers may not be
+    // persisted long-term.
+    if (auto ledger = getLedgerByHash(ledgerHash))
+        return ledger->info().closeTime;
+
     auto nodeObject = app_.getNodeStore().fetchNodeObject(ledgerHash, index);
     if (nodeObject && (nodeObject->getData().size() >= 120))
     {
@@ -1807,6 +1852,85 @@ LedgerMaster::getLedgerByHash(uint256 const& hash)
     return {};
 }
 
+std::shared_ptr<Ledger const>
+LedgerMaster::getClosestFullyWiredLedger(
+    std::shared_ptr<Ledger const> const& targetLedger)
+{
+    if (!targetLedger)
+        return {};
+
+    std::vector<std::shared_ptr<Ledger const>> candidates;
+    {
+        std::lock_guard lock(m_mutex);
+        candidates.reserve(mRetainedLedgers.size() + 3);
+        for (auto const& ledger : mRetainedLedgers)
+            candidates.push_back(ledger);
+        if (auto const closed = mClosedLedger.get())
+            candidates.push_back(closed);
+        if (auto const valid = mValidLedger.get())
+            candidates.push_back(valid);
+        if (mPubLedger)
+            candidates.push_back(mPubLedger);
+    }
+
+    auto const targetSeq = targetLedger->info().seq;
+    auto const targetHash = targetLedger->info().hash;
+
+    std::shared_ptr<Ledger const> best;
+    auto bestDistance = std::numeric_limits<std::uint32_t>::max();
+
+    for (auto const& candidate : candidates)
+    {
+        if (!candidate || !candidate->isFullyWired())
+            continue;
+
+        if (candidate->info().hash == targetHash)
+            continue;
+
+        bool sameChain = false;
+        try
+        {
+            if (candidate->info().seq < targetSeq)
+            {
+                if (auto const hash = hashOfSeq(
+                        *targetLedger, candidate->info().seq, m_journal);
+                    hash && *hash == candidate->info().hash)
+                {
+                    sameChain = true;
+                }
+            }
+            else if (candidate->info().seq > targetSeq)
+            {
+                if (auto const hash =
+                        hashOfSeq(*candidate, targetSeq, m_journal);
+                    hash && *hash == targetHash)
+                {
+                    sameChain = true;
+                }
+            }
+        }
+        catch (std::exception const&)
+        {
+            sameChain = false;
+        }
+
+        if (!sameChain)
+            continue;
+
+        auto const distance = candidate->info().seq < targetSeq
+            ? targetSeq - candidate->info().seq
+            : candidate->info().seq - targetSeq;
+
+        if (!best || distance < bestDistance)
+        {
+            best = candidate;
+            bestDistance = distance;
+        }
+    }
+
+    return best;
+}
+
 void
 LedgerMaster::setLedgerRangePresent(
     std::uint32_t minV,

src/xrpld/app/misc/NetworkOPs.cpp

@@ -897,9 +897,11 @@ NetworkOPsImp::setHeartbeatTimer()
         heartbeatTimer_,
         mConsensus.parms().ledgerGRANULARITY,
         [this]() {
-            m_job_queue.addJob(jtNETOP_TIMER, "NetOPs.heartbeat", [this]() {
-                processHeartbeatTimer();
-            });
+            // Run the heartbeat directly on the io_service thread instead
+            // of posting to the JobQueue.  This prevents heavy RPC load
+            // from starving the consensus heartbeat timer — the io_service
+            // thread pool is independent of the JobQueue worker pool.
+            processHeartbeatTimer();
         },
         [this]() { setHeartbeatTimer(); });
 }
@@ -939,66 +941,82 @@ NetworkOPsImp::processHeartbeatTimer()
     RclConsensusLogger clog(
         "Heartbeat Timer", mConsensus.validating(), m_journal);
     {
-        std::unique_lock lock{app_.getMasterMutex()};
+        // Use try_to_lock so the heartbeat never blocks on masterMutex.
+        // If apply() or another operation is holding it, skip the non-critical
+        // peer/mode checks and proceed directly to timerEntry() — ensuring
+        // consensus timing is never delayed by mutex contention.
+        std::unique_lock lock{app_.getMasterMutex(), std::try_to_lock};
 
-        // VFALCO NOTE This is for diagnosing a crash on exit
-        LoadManager& mgr(app_.getLoadManager());
-        mgr.resetDeadlockDetector();
-
-        std::size_t const numPeers = app_.overlay().size();
-
-        // do we have sufficient peers? If not, we are disconnected.
-        if (numPeers < minPeerCount_)
+        if (lock.owns_lock())
         {
-            if (mMode != OperatingMode::DISCONNECTED)
+            // VFALCO NOTE This is for diagnosing a crash on exit
+            LoadManager& mgr(app_.getLoadManager());
+            mgr.resetDeadlockDetector();
+
+            std::size_t const numPeers = app_.overlay().size();
+
+            // do we have sufficient peers? If not, we are disconnected.
+            if (numPeers < minPeerCount_)
             {
-                setMode(OperatingMode::DISCONNECTED);
-                std::stringstream ss;
-                ss << "Node count (" << numPeers << ") has fallen "
-                   << "below required minimum (" << minPeerCount_ << ").";
-                JLOG(m_journal.warn()) << ss.str();
-                CLOG(clog.ss()) << "set mode to DISCONNECTED: " << ss.str();
+                if (mMode != OperatingMode::DISCONNECTED)
+                {
+                    setMode(OperatingMode::DISCONNECTED);
+                    std::stringstream ss;
+                    ss << "Node count (" << numPeers << ") has fallen "
+                       << "below required minimum (" << minPeerCount_ << ").";
+                    JLOG(m_journal.warn()) << ss.str();
+                    CLOG(clog.ss()) << "set mode to DISCONNECTED: " << ss.str();
+                }
+                else
+                {
+                    CLOG(clog.ss())
+                        << "already DISCONNECTED. too few peers (" << numPeers
+                        << "), need at least " << minPeerCount_;
+                }
+
+                // MasterMutex lock need not be held to call
+                // setHeartbeatTimer()
+                lock.unlock();
+                // We do not call mConsensus.timerEntry until there are
+                // enough peers providing meaningful inputs to consensus
+                setHeartbeatTimer();
+
+                return;
             }
-            else
+
+            if (mMode == OperatingMode::DISCONNECTED)
+            {
+                setMode(OperatingMode::CONNECTED);
+                JLOG(m_journal.info())
+                    << "Node count (" << numPeers << ") is sufficient.";
+                CLOG(clog.ss()) << "setting mode to CONNECTED based on "
+                                << numPeers << " peers. ";
+            }
+
+            // Check if the last validated ledger forces a change between
+            // these states.
+            auto origMode = mMode.load();
+            CLOG(clog.ss()) << "mode: " << strOperatingMode(origMode, true);
+            if (mMode == OperatingMode::SYNCING)
+                setMode(OperatingMode::SYNCING);
+            else if (mMode == OperatingMode::CONNECTED)
+                setMode(OperatingMode::CONNECTED);
+            auto newMode = mMode.load();
+            if (origMode != newMode)
             {
                 CLOG(clog.ss())
-                    << "already DISCONNECTED. too few peers (" << numPeers
-                    << "), need at least " << minPeerCount_;
+                    << ", changing to " << strOperatingMode(newMode, true);
             }
-
-            // MasterMutex lock need not be held to call setHeartbeatTimer()
-            lock.unlock();
-            // We do not call mConsensus.timerEntry until there are enough
-            // peers providing meaningful inputs to consensus
-            setHeartbeatTimer();
-
-            return;
+            CLOG(clog.ss()) << ". ";
         }
-
-        if (mMode == OperatingMode::DISCONNECTED)
-        {
-            setMode(OperatingMode::CONNECTED);
-            JLOG(m_journal.info())
-                << "Node count (" << numPeers << ") is sufficient.";
-            CLOG(clog.ss()) << "setting mode to CONNECTED based on " << numPeers
-                            << " peers. ";
-        }
-
-        // Check if the last validated ledger forces a change between these
-        // states.
-        auto origMode = mMode.load();
-        CLOG(clog.ss()) << "mode: " << strOperatingMode(origMode, true);
-        if (mMode == OperatingMode::SYNCING)
-            setMode(OperatingMode::SYNCING);
-        else if (mMode == OperatingMode::CONNECTED)
-            setMode(OperatingMode::CONNECTED);
-        auto newMode = mMode.load();
-        if (origMode != newMode)
+        else
         {
+            JLOG(m_journal.debug())
+                << "Heartbeat: masterMutex contended, skipping "
+                   "peer/mode checks";
             CLOG(clog.ss())
-                << ", changing to " << strOperatingMode(newMode, true);
+                << "masterMutex contended, skipping peer/mode checks. ";
         }
-        CLOG(clog.ss()) << ". ";
     }
 
     mConsensus.timerEntry(app_.timeKeeper().closeTime(), clog.ss());

src/xrpld/app/misc/SHAMapStoreImp.cpp

@@ -116,6 +116,18 @@ SHAMapStoreImp::SHAMapStoreImp(
     }
 
     get_if_exists(section, "online_delete", deleteInterval_);
+    isMemoryBackend_ = boost::iequals(get(section, "type"), "rwdb");
+
+    // For RWDB, default online_delete to ledger_history only if user did not
+    // explicitly set online_delete.  Clamp to the minimum so an implicit
+    // value never triggers the "online_delete must be at least …" throw.
+    if (isMemoryBackend_ && deleteInterval_ == 0)
+    {
+        auto const minInterval = config.standalone()
+            ? minimumDeletionIntervalSA_
+            : minimumDeletionInterval_;
+        deleteInterval_ = std::max(config.LEDGER_HISTORY, minInterval);
+    }
 
     if (deleteInterval_)
     {
@@ -154,7 +166,7 @@ SHAMapStoreImp::SHAMapStoreImp(
         }
 
         state_db_.init(config, dbName_);
-        if (!config.mem_backend())
+        if (!isMemoryBackend_)
             dbPaths();
     }
 }
@@ -325,64 +337,137 @@ SHAMapStoreImp::run()
             if (healthWait() == stopping)
                 return;
 
-            JLOG(journal_.debug()) << "copying ledger " << validatedSeq;
-            std::uint64_t nodeCount = 0;
-
-            try
+            if (isMemoryBackend_)
             {
-                validatedLedger->stateMap().snapShot(false)->visitNodes(
-                    std::bind(
-                        &SHAMapStoreImp::copyNode,
-                        this,
-                        std::ref(nodeCount),
-                        std::placeholders::_1));
+                // For RWDB: copy only the current validated ledger's live
+                // state nodes into a fresh backend that is not yet shared,
+                // avoiding both exclusive-lock contention on the live
+                // writable backend AND stale-node accumulation.
+                //
+                // copyArchiveTo would carry forward ALL archive entries
+                // (including stale nodes from older ledger versions that
+                // were promoted via fetch duplication), causing unbounded
+                // memory growth across rotation cycles.
+                JLOG(journal_.debug()) << "RWDB: copying live state for rotation";
+                auto newBackend = makeBackendRotating();
+                std::uint64_t nodeCount = 0;
+                bool aborted = false;
+
+                try
+                {
+                    validatedLedger->stateMap().snapShot(false)->visitNodes(
+                        [&](SHAMapTreeNode& node) -> bool {
+                            auto const hash = node.getHash().as_uint256();
+                            // Fetch the NodeObject from the rotating DB
+                            // (checks writable then archive) and store it
+                            // directly in the new unshared backend.
+                            auto obj = dbRotating_->fetchNodeObject(
+                                hash,
+                                0,
+                                NodeStore::FetchType::synchronous,
+                                false);
+                            if (obj)
+                                newBackend->store(obj);
+
+                            if ((++nodeCount % checkHealthInterval_) == 0)
+                            {
+                                if (healthWait() == stopping)
+                                {
+                                    aborted = true;
+                                    return false;
+                                }
+                            }
+                            return true;
+                        });
+                }
+                catch (SHAMapMissingNode const& e)
+                {
+                    JLOG(journal_.error())
+                        << "Missing node while copying state before rotate: "
+                        << e.what();
+                    continue;
+                }
+
+                if (aborted)
+                    return;
+                JLOG(journal_.debug())
+                    << "RWDB: copied " << nodeCount << " live nodes";
+
+                ledgerMaster_->clearLedgerCachePrior(validatedSeq);
+                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;
+                        state_db_.setState(savedState);
+                        ledgerMaster_->clearLedgerCachePrior(validatedSeq);
+                    });
+
+                JLOG(journal_.warn()) << "finished rotation " << validatedSeq;
             }
-            catch (SHAMapMissingNode const& e)
+            else
             {
-                JLOG(journal_.error())
-                    << "Missing node while copying ledger before rotate: "
-                    << e.what();
-                continue;
+                JLOG(journal_.debug()) << "copying ledger " << validatedSeq;
+                std::uint64_t nodeCount = 0;
+
+                try
+                {
+                    validatedLedger->stateMap().snapShot(false)->visitNodes(
+                        std::bind(
+                            &SHAMapStoreImp::copyNode,
+                            this,
+                            std::ref(nodeCount),
+                            std::placeholders::_1));
+                }
+                catch (SHAMapMissingNode const& e)
+                {
+                    JLOG(journal_.error())
+                        << "Missing node while copying ledger before rotate: "
+                        << e.what();
+                    continue;
+                }
+
+                if (healthWait() == stopping)
+                    return;
+                JLOG(journal_.debug()) << "copied ledger " << validatedSeq
+                                       << " nodecount " << nodeCount;
+
+                JLOG(journal_.debug()) << "freshening caches";
+                freshenCaches();
+                if (healthWait() == stopping)
+                    return;
+                JLOG(journal_.debug()) << validatedSeq << " freshened caches";
+
+                JLOG(journal_.trace()) << "Making a new backend";
+                auto newBackend = makeBackendRotating();
+                JLOG(journal_.debug())
+                    << validatedSeq << " new backend " << newBackend->getName();
+
+                clearCaches(validatedSeq);
+                if (healthWait() == stopping)
+                    return;
+
+                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;
+                        state_db_.setState(savedState);
+                        clearCaches(validatedSeq);
+                    });
+
+                JLOG(journal_.warn()) << "finished rotation " << validatedSeq;
             }
-
-            if (healthWait() == stopping)
-                return;
-            // Only log if we completed without a "health" abort
-            JLOG(journal_.debug()) << "copied ledger " << validatedSeq
-                                   << " nodecount " << nodeCount;
-
-            JLOG(journal_.debug()) << "freshening caches";
-            freshenCaches();
-            if (healthWait() == stopping)
-                return;
-            // Only log if we completed without a "health" abort
-            JLOG(journal_.debug()) << validatedSeq << " freshened caches";
-
-            JLOG(journal_.debug()) << "Making a new backend";
-            auto newBackend = makeBackendRotating();
-            JLOG(journal_.debug())
-                << validatedSeq << " new backend " << newBackend->getName();
-
-            clearCaches(validatedSeq);
-            if (healthWait() == stopping)
-                return;
-
-            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;
-                    state_db_.setState(savedState);
-
-                    clearCaches(validatedSeq);
-                });
-
-            JLOG(journal_.warn()) << "finished rotation " << validatedSeq;
         }
     }
 }

src/xrpld/app/misc/SHAMapStoreImp.h

@@ -101,6 +101,7 @@ private:
 
     std::uint32_t deleteInterval_ = 0;
     bool advisoryDelete_ = false;
+    bool isMemoryBackend_ = false;
     std::uint32_t deleteBatch_ = 100;
     std::chrono::milliseconds backOff_{100};
     std::chrono::seconds ageThreshold_{60};

src/xrpld/nodestore/DatabaseRotating.h

@@ -55,6 +55,15 @@ public:
         std::function<void(
             std::string const& writableName,
             std::string const& archiveName)> const& f) = 0;
+
+    /** Populate @a dest with every object in the archive backend.
+
+        Used by in-memory (RWDB) backends to pre-populate a new writable
+        backend before rotation, avoiding per-node write-lock contention on
+        the live writable backend.  @a dest must not yet be shared.
+    */
+    virtual void
+    copyArchiveTo(Backend& dest) = 0;
 };
 
 }  // namespace NodeStore

src/xrpld/nodestore/backend/RWDBFactory.cpp

@@ -3,12 +3,16 @@
 #include <xrpld/nodestore/detail/DecodedBlob.h>
 #include <xrpld/nodestore/detail/EncodedBlob.h>
 #include <xrpld/nodestore/detail/codec.h>
+#include <xrpl/basics/ReaderPreferringSharedMutex.h>
 #include <xrpl/basics/contract.h>
 #include <boost/beast/core/string.hpp>
 #include <boost/core/ignore_unused.hpp>
 #include <boost/unordered/concurrent_flat_map.hpp>
+#include <cstdlib>
 #include <memory>
 #include <mutex>
+#include <shared_mutex>
+#include <string_view>
 
 namespace ripple {
 namespace NodeStore {
@@ -34,8 +38,7 @@ private:
     using DataStore =
         std::map<uint256, std::vector<std::uint8_t>>;  // Store compressed blob
                                                        // data
-    mutable std::recursive_mutex
-        mutex_;  // Only needed for std::map implementation
+    mutable reader_preferring_shared_mutex mutex_;
 
     DataStore table_;
 
@@ -65,7 +68,7 @@ public:
     void
     open(bool createIfMissing) override
     {
-        std::lock_guard lock(mutex_);
+        std::unique_lock lock(mutex_);
         if (isOpen_)
             Throw<std::runtime_error>("already open");
         isOpen_ = true;
@@ -74,26 +77,44 @@ public:
     bool
     isOpen() override
     {
+        std::shared_lock lock(mutex_);
         return isOpen_;
     }
 
     void
     close() override
     {
-        std::lock_guard lock(mutex_);
-        table_.clear();
-        isOpen_ = false;
+        DataStore old;
+        {
+            std::unique_lock lock(mutex_);
+            isOpen_ = false;
+            old.swap(table_);  // O(1) swap; release lock before destructor runs
+        }
+        // 'old' is now destroyed outside the lock — no fetch() can be
+        // blocked by the (potentially millions-of-entries) map destructor.
+    }
+
+    static bool
+    nullMode()
+    {
+        static bool const v = [] {
+            char const* e = std::getenv("XAHAU_RWDB_NULL");
+            return e && *e && std::string_view{e} != "0";
+        }();
+        return v;
     }
 
     Status
     fetch(void const* key, std::shared_ptr<NodeObject>* pObject) override
     {
-        if (!isOpen_)
+        if (nullMode())
             return notFound;
 
         uint256 const hash(uint256::fromVoid(key));
 
-        std::lock_guard lock(mutex_);
+        std::shared_lock lock(mutex_);
+        if (!isOpen_)
+            return notFound;
         auto it = table_.find(hash);
         if (it == table_.end())
             return notFound;
@@ -134,6 +155,17 @@ public:
         if (!object)
             return;
 
+        if (nullMode())
+            return;
+
+        static bool const discardHotAccountNode = [] {
+            char const* v = std::getenv("XAHAU_RWDB_DISCARD_HOT_ACCOUNT_NODE");
+            return v && *v && std::string_view{v} != "0";
+        }();
+
+        if (discardHotAccountNode && object->getType() == hotACCOUNT_NODE)
+            return;
+
         EncodedBlob encoded(object);
         nudb::detail::buffer bf;
         auto const result =
@@ -162,10 +194,9 @@ public:
     void
     for_each(std::function<void(std::shared_ptr<NodeObject>)> f) override
     {
+        std::shared_lock lock(mutex_);
         if (!isOpen_)
             return;
-
-        std::lock_guard lock(mutex_);
         for (const auto& entry : table_)
         {
             nudb::detail::buffer bf;

src/xrpld/nodestore/detail/DatabaseRotatingImp.cpp

@@ -44,6 +44,21 @@ DatabaseRotatingImp::DatabaseRotatingImp(
         fdRequired_ += archiveBackend_->fdRequired();
 }
 
+void
+DatabaseRotatingImp::copyArchiveTo(Backend& dest)
+{
+    // Snapshot the archive backend pointer under lock, then iterate it
+    // outside the lock.  dest is not yet shared so its store() calls are
+    // uncontested — no live-backend write-lock contention.
+    auto archive = [&] {
+        std::lock_guard const lock(mutex_);
+        return archiveBackend_;
+    }();
+
+    archive->for_each(
+        [&](std::shared_ptr<NodeObject> obj) { dest.store(obj); });
+}
+
 void
 DatabaseRotatingImp::rotate(
     std::unique_ptr<NodeStore::Backend>&& newBackend,
@@ -111,8 +126,11 @@ DatabaseRotatingImp::rotate(
         // Execute the lambda
         ensurePinnedLedgersInWritable();
 
-        // Now it's safe to mark the archive backend for deletion
-        archiveBackend_->setDeletePath();
+        // Do NOT call setDeletePath() inside this lock.  For in-memory
+        // backends, setDeletePath() calls close() which destructs the entire
+        // table_ map (millions of shared_ptr<NodeObject> ref-count decrements)
+        // while the lock is held, blocking every concurrent fetchNodeObject
+        // call for several seconds and starving consensus reads.
         oldArchiveBackend = std::move(archiveBackend_);
 
         // Complete the rotation
@@ -122,6 +140,9 @@ DatabaseRotatingImp::rotate(
         writableBackend_ = std::move(newBackend);
     }
 
+    // Lock released — clear the old archive now without blocking fetches.
+    oldArchiveBackend->setDeletePath();
+
     f(newWritableBackendName, newArchiveBackendName);
 }
 

src/xrpld/nodestore/detail/DatabaseRotatingImp.h

@@ -51,6 +51,9 @@ public:
         stop();
     }
 
+    void
+    copyArchiveTo(Backend& dest) override;
+
     void
     rotate(
         std::unique_ptr<NodeStore::Backend>&& newBackend,

src/xrpld/overlay/detail/PeerImp.cpp

@@ -32,11 +32,14 @@
 #include <xrpld/overlay/detail/PeerImp.h>
 #include <xrpld/overlay/detail/Tuning.h>
 #include <xrpld/perflog/PerfLog.h>
+#include <xrpld/shamap/Family.h>
+#include <xrpld/shamap/SHAMapTreeNode.h>
 #include <xrpl/basics/UptimeClock.h>
 #include <xrpl/basics/base64.h>
 #include <xrpl/basics/random.h>
 #include <xrpl/basics/safe_cast.h>
 #include <xrpl/beast/core/LexicalCast.h>
+#include <xrpl/protocol/HashPrefix.h>
 #include <xrpl/protocol/digest.h>
 
 #include <boost/algorithm/string/predicate.hpp>
@@ -2464,13 +2467,50 @@ PeerImp::onMessage(std::shared_ptr<protocol::TMGetObjectByHash> const& m)
                 //             need to inject the NodeStore interfaces.
                 std::uint32_t seq{obj.has_ledgerseq() ? obj.ledgerseq() : 0};
                 auto nodeObject{app_.getNodeStore().fetchNodeObject(hash, seq)};
+
+                void const* dataPtr = nullptr;
+                std::size_t dataSize = 0;
+                Blob treeBlob;
+
                 if (nodeObject)
+                {
+                    dataPtr = nodeObject->getData().data();
+                    dataSize = nodeObject->getData().size();
+                }
+                else if (
+                    auto treeNode =
+                        app_.getNodeFamily().getTreeNodeCache()->fetch(hash))
+                {
+                    // SHAMap tree node fallback — works for state/tx nodes
+                    // held via the retained Ledgers' SHAMap inner nodes.
+                    Serializer s;
+                    treeNode->serializeWithPrefix(s);
+                    treeBlob = std::move(s.modData());
+                    dataPtr = treeBlob.data();
+                    dataSize = treeBlob.size();
+                }
+                else if (packet.type() == protocol::TMGetObjectByHash::otLEDGER)
+                {
+                    // Ledger header fallback — look up by hash in the
+                    // in-memory ledger set and serialize the header in the
+                    // same wire format used by the node store.
+                    if (auto ledger =
+                            app_.getLedgerMaster().getLedgerByHash(hash))
+                    {
+                        Serializer s(sizeof(LedgerInfo) + 4);
+                        s.add32(HashPrefix::ledgerMaster);
+                        addRaw(ledger->info(), s);
+                        treeBlob = std::move(s.modData());
+                        dataPtr = treeBlob.data();
+                        dataSize = treeBlob.size();
+                    }
+                }
+
+                if (dataPtr)
                 {
                     protocol::TMIndexedObject& newObj = *reply.add_objects();
                     newObj.set_hash(hash.begin(), hash.size());
-                    newObj.set_data(
-                        &nodeObject->getData().front(),
-                        nodeObject->getData().size());
+                    newObj.set_data(dataPtr, dataSize);
 
                     if (obj.has_nodeid())
                         newObj.set_index(obj.nodeid());

src/xrpld/shamap/detail/SHAMapSync.cpp

@@ -21,8 +21,25 @@
 #include <xrpld/shamap/SHAMapSyncFilter.h>
 #include <xrpl/basics/random.h>
 
+#include <cstdlib>
+#include <string_view>
+
 namespace ripple {
 
+namespace {
+
+bool
+useFullBelowCache()
+{
+    static bool const use = [] {
+        char const* e = std::getenv("XAHAU_RWDB_NULL");
+        return !(e && *e && std::string_view{e} != "0");
+    }();
+    return use;
+}
+
+}  // namespace
+
 void
 SHAMap::visitLeaves(
     std::function<void(boost::intrusive_ptr<SHAMapItem const> const&
@@ -191,7 +208,7 @@ SHAMap::gmn_ProcessNodes(MissingNodes& mn, MissingNodes::StackEntry& se)
             fullBelow = false;
         }
         else if (
-            !backed_ ||
+            !backed_ || !useFullBelowCache() ||
             !f_.getFullBelowCache()->touch_if_exists(childHash.as_uint256()))
         {
             bool pending = false;
@@ -228,7 +245,9 @@ SHAMap::gmn_ProcessNodes(MissingNodes& mn, MissingNodes::StackEntry& se)
             }
             else if (
                 d->isInner() &&
-                !static_cast<SHAMapInnerNode*>(d)->isFullBelow(mn.generation_))
+                (!useFullBelowCache() ||
+                 !static_cast<SHAMapInnerNode*>(d)->isFullBelow(
+                     mn.generation_)))
             {
                 mn.stack_.push(se);
 
@@ -248,7 +267,7 @@ SHAMap::gmn_ProcessNodes(MissingNodes& mn, MissingNodes::StackEntry& se)
     if (fullBelow)
     {  // No partial node encountered below this node
         node->setFullBelowGen(mn.generation_);
-        if (backed_)
+        if (backed_ && useFullBelowCache())
         {
             f_.getFullBelowCache()->insert(node->getHash().as_uint256());
         }
@@ -326,8 +345,9 @@ SHAMap::getMissingNodes(int max, SHAMapSyncFilter* filter)
         f_.getFullBelowCache()->getGeneration());
 
     if (!root_->isInner() ||
-        std::static_pointer_cast<SHAMapInnerNode>(root_)->isFullBelow(
-            mn.generation_))
+        (useFullBelowCache() &&
+         std::static_pointer_cast<SHAMapInnerNode>(root_)->isFullBelow(
+             mn.generation_)))
     {
         clearSynching();
         return std::move(mn.missingNodes_);
@@ -397,7 +417,8 @@ SHAMap::getMissingNodes(int max, SHAMapSyncFilter* filter)
             {
                 // Recheck nodes we could not finish before
                 for (auto const& [innerNode, nodeId] : mn.resumes_)
-                    if (!innerNode->isFullBelow(mn.generation_))
+                    if (!useFullBelowCache() ||
+                        !innerNode->isFullBelow(mn.generation_))
                         mn.stack_.push(std::make_tuple(
                             innerNode, nodeId, rand_int(255), 0, true));
 
@@ -592,7 +613,8 @@ SHAMap::addKnownNode(
     auto iNode = root_.get();
 
     while (iNode->isInner() &&
-           !static_cast<SHAMapInnerNode*>(iNode)->isFullBelow(generation) &&
+           (!useFullBelowCache() ||
+            !static_cast<SHAMapInnerNode*>(iNode)->isFullBelow(generation)) &&
            (iNodeID.getDepth() < node.getDepth()))
     {
         int branch = selectBranch(iNodeID, node.getNodeID());
@@ -605,7 +627,8 @@ SHAMap::addKnownNode(
         }
 
         auto childHash = inner->getChildHash(branch);
-        if (f_.getFullBelowCache()->touch_if_exists(childHash.as_uint256()))
+        if (useFullBelowCache() &&
+            f_.getFullBelowCache()->touch_if_exists(childHash.as_uint256()))
         {
             return SHAMapAddNode::duplicate();
         }