Merge branch 'develop' into ximinez/online-delete-gaps

2026-07-28 01:20:32 +00:00 · 2026-06-17 15:01:23 -04:00
parent 1116dbfd0e 480676d0bf
commit dd4eca22b7
42 changed files with 929 additions and 619 deletions
--- a/src/libxrpl/ledger/BookListeners.cpp
+++ b/src/libxrpl/ledger/BookListeners.cpp
@@ -1,55 +0,0 @@
-#include <xrpl/ledger/BookListeners.h>
-
-#include <xrpl/basics/UnorderedContainers.h>
-#include <xrpl/json/json_value.h>
-#include <xrpl/protocol/MultiApiJson.h>
-#include <xrpl/server/InfoSub.h>
-
-#include <cstdint>
-#include <mutex>
-
-namespace xrpl {
-
-void
-BookListeners::addSubscriber(InfoSub::ref sub)
-{
-    std::scoped_lock const sl(lock_);
-    listeners_[sub->getSeq()] = sub;
-}
-
-void
-BookListeners::removeSubscriber(std::uint64_t seq)
-{
-    std::scoped_lock const sl(lock_);
-    listeners_.erase(seq);
-}
-
-void
-BookListeners::publish(MultiApiJson const& jvObj, hash_set<std::uint64_t>& havePublished)
-{
-    std::scoped_lock const sl(lock_);
-    auto it = listeners_.cbegin();
-
-    while (it != listeners_.cend())
-    {
-        InfoSub::pointer p = it->second.lock();
-
-        if (p)
-        {
-            // Only publish jvObj if this is the first occurrence
-            if (havePublished.emplace(p->getSeq()).second)
-            {
-                jvObj.visit(
-                    p->getApiVersion(),  //
-                    [&](json::Value const& jv) { p->send(jv, true); });
-            }
-            ++it;
-        }
-        else
-        {
-            it = listeners_.erase(it);
-        }
-    }
-}
-
-}  // namespace xrpl
--- a/src/libxrpl/ledger/helpers/PaymentChannelHelpers.cpp
+++ b/src/libxrpl/ledger/helpers/PaymentChannelHelpers.cpp
@@ -5,13 +5,20 @@
 #include <xrpl/beast/utility/Journal.h>
 #include <xrpl/beast/utility/instrumentation.h>
 #include <xrpl/ledger/ApplyView.h>
+#include <xrpl/ledger/View.h>
 #include <xrpl/ledger/helpers/AccountRootHelpers.h>
 #include <xrpl/protocol/AccountID.h>
+#include <xrpl/protocol/Feature.h>
 #include <xrpl/protocol/Indexes.h>
 #include <xrpl/protocol/SField.h>
 #include <xrpl/protocol/STLedgerEntry.h>
 #include <xrpl/protocol/TER.h>

+#include <algorithm>
+#include <cstdint>
+#include <limits>
+#include <optional>
+
 namespace xrpl {

 TER
@@ -59,4 +66,28 @@ closeChannel(SLE::ref slep, ApplyView& view, uint256 const& key, beast::Journal
    return tesSUCCESS;
 }

+uint32_t
+saturatingAdd(Rules const& rules, uint32_t const lhs, uint32_t const rhs)
+{
+    if (rules.enabled(fixCleanup3_2_0))
+    {
+        static constexpr auto kUint32Max =
+            static_cast<uint64_t>(std::numeric_limits<uint32_t>::max());
+        uint64_t const saturatedResult = std::min(uint64_t{lhs} + rhs, kUint32Max);
+        return static_cast<uint32_t>(saturatedResult);
+    }
+
+    return lhs + rhs;
+}
+
+bool
+isChannelExpired(ApplyView const& view, std::optional<uint32_t> timeField)
+{
+    if (!timeField)
+        return false;
+    if (view.rules().enabled(fixCleanup3_2_0))
+        return after(view.header().parentCloseTime, *timeField);
+    return view.header().parentCloseTime.time_since_epoch().count() >= *timeField;
+}
+
 }  // namespace xrpl
--- a/src/libxrpl/nodestore/ManagerImp.cpp
+++ b/src/libxrpl/nodestore/ManagerImp.cpp
@@ -45,8 +45,10 @@ ManagerImp::missingBackend()
 // the Factory classes is an undefined behaviour.
 void
 registerNuDBFactory(Manager& manager);
+#if XRPL_ROCKSDB_AVAILABLE
 void
 registerRocksDBFactory(Manager& manager);
+#endif
 void
 registerNullFactory(Manager& manager);
 void
@@ -55,7 +57,9 @@ registerMemoryFactory(Manager& manager);
 ManagerImp::ManagerImp()
 {
    registerNuDBFactory(*this);
+#if XRPL_ROCKSDB_AVAILABLE
    registerRocksDBFactory(*this);
+#endif
    registerNullFactory(*this);
    registerMemoryFactory(*this);
 }
--- a/src/libxrpl/nodestore/backend/RocksDBFactory.cpp
+++ b/src/libxrpl/nodestore/backend/RocksDBFactory.cpp
@@ -1,13 +1,23 @@
+#if XRPL_ROCKSDB_AVAILABLE
+#include <xrpl/basics/ByteUtilities.h>
 #include <xrpl/basics/Log.h>
 #include <xrpl/basics/base_uint.h>
+#include <xrpl/basics/contract.h>
+#include <xrpl/basics/safe_cast.h>
+#include <xrpl/beast/core/CurrentThreadName.h>
 #include <xrpl/beast/utility/Journal.h>
 #include <xrpl/beast/utility/instrumentation.h>
 #include <xrpl/config/BasicConfig.h>
 #include <xrpl/config/Constants.h>
 #include <xrpl/nodestore/Backend.h>
+#include <xrpl/nodestore/Factory.h>
+#include <xrpl/nodestore/Manager.h>
 #include <xrpl/nodestore/NodeObject.h>
 #include <xrpl/nodestore/Scheduler.h>
 #include <xrpl/nodestore/Types.h>
+#include <xrpl/nodestore/detail/BatchWriter.h>
+#include <xrpl/nodestore/detail/DecodedBlob.h>
+#include <xrpl/nodestore/detail/EncodedBlob.h>

 #include <boost/filesystem/operations.hpp>
 #include <boost/filesystem/path.hpp>
@@ -25,26 +35,14 @@
 #include <rocksdb/table.h>
 #include <rocksdb/write_batch.h>

+#include <atomic>
 #include <bit>
 #include <cstddef>
 #include <functional>
+#include <memory>
 #include <stdexcept>
 #include <string>

-#if XRPL_ROCKSDB_AVAILABLE
-#include <xrpl/basics/ByteUtilities.h>
-#include <xrpl/basics/contract.h>
-#include <xrpl/basics/safe_cast.h>
-#include <xrpl/beast/core/CurrentThreadName.h>
-#include <xrpl/nodestore/Factory.h>
-#include <xrpl/nodestore/Manager.h>
-#include <xrpl/nodestore/detail/BatchWriter.h>
-#include <xrpl/nodestore/detail/DecodedBlob.h>
-#include <xrpl/nodestore/detail/EncodedBlob.h>
-
-#include <atomic>
-#include <memory>
-
 namespace xrpl::NodeStore {

 class RocksDBEnv : public rocksdb::EnvWrapper
--- a/src/libxrpl/protocol/BuildInfo.cpp
+++ b/src/libxrpl/protocol/BuildInfo.cpp
@@ -23,7 +23,7 @@ namespace {
 //------------------------------------------------------------------------------
 // clang-format off
 // NOLINTNEXTLINE(readability-identifier-naming)
-char const* const versionString = "3.2.0-rc3"
+char const* const versionString = "3.3.0-b0"
    // clang-format on
    ;

--- a/src/libxrpl/server/InfoSub.cpp
+++ b/src/libxrpl/server/InfoSub.cpp
@@ -1,15 +1,47 @@
 #include <xrpl/server/InfoSub.h>

+#include <xrpl/basics/Log.h>
+#include <xrpl/beast/utility/Journal.h>
 #include <xrpl/beast/utility/instrumentation.h>
 #include <xrpl/protocol/AccountID.h>
+#include <xrpl/protocol/Book.h>
 #include <xrpl/resource/Consumer.h>

 #include <cstdint>
+#include <exception>
 #include <memory>
 #include <mutex>

 namespace xrpl {

+namespace {
+
+// Wraps a Source teardown call so that an exception from one cleanup
+// step does not prevent the subsequent steps from running. Source methods
+// acquire a lock and can throw std::system_error; a throw out of ~InfoSub
+// during stack unwinding would terminate the process. Failures are
+// reported through the Source's Journal so they reach the configured log
+// sinks; JLOG itself cannot throw, so the noexcept guarantee holds.
+template <typename F>
+void
+safeUnsub(std::uint64_t seq, F&& f, beast::Journal j) noexcept
+{
+    try
+    {
+        f();
+    }
+    catch (std::exception const& e)
+    {
+        JLOG(j.warn()) << "~InfoSub[seq=" << seq << "]: cleanup step failed: " << e.what();
+    }
+    catch (...)
+    {
+        JLOG(j.warn()) << "~InfoSub[seq=" << seq << "]: cleanup step failed: unknown exception";
+    }
+}
+
+}  // namespace
+
 // This is the primary interface into the "client" portion of the program.
 // Code that wants to do normal operations on the network such as
 // creating and monitoring accounts, creating transactions, and so on
@@ -32,25 +64,44 @@ InfoSub::InfoSub(Source& source, Consumer consumer)

 InfoSub::~InfoSub()
 {
-    source_.unsubTransactions(seq_);
-    source_.unsubRTTransactions(seq_);
-    source_.unsubLedger(seq_);
-    source_.unsubManifests(seq_);
-    source_.unsubServer(seq_);
-    source_.unsubValidations(seq_);
-    source_.unsubPeerStatus(seq_);
-    source_.unsubConsensus(seq_);
+    // Each Source teardown call below acquires a server-side lock and
+    // can throw. Wrap each independent call so partial failure does not
+    // skip the remaining teardown steps.
+
+    auto const& j = source_.journal();
+
+    safeUnsub(seq_, [&] { source_.unsubTransactions(seq_); }, j);
+    safeUnsub(seq_, [&] { source_.unsubRTTransactions(seq_); }, j);
+    safeUnsub(seq_, [&] { source_.unsubLedger(seq_); }, j);
+    safeUnsub(seq_, [&] { source_.unsubManifests(seq_); }, j);
+    safeUnsub(seq_, [&] { source_.unsubServer(seq_); }, j);
+    safeUnsub(seq_, [&] { source_.unsubValidations(seq_); }, j);
+    safeUnsub(seq_, [&] { source_.unsubPeerStatus(seq_); }, j);
+    safeUnsub(seq_, [&] { source_.unsubConsensus(seq_); }, j);

    // Use the internal unsubscribe so that it won't call
    // back to us and modify its own parameter
    if (!realTimeSubscriptions_.empty())
-        source_.unsubAccountInternal(seq_, realTimeSubscriptions_, true);
+    {
+        safeUnsub(
+            seq_, [&] { source_.unsubAccountInternal(seq_, realTimeSubscriptions_, true); }, j);
+    }

    if (!normalSubscriptions_.empty())
-        source_.unsubAccountInternal(seq_, normalSubscriptions_, false);
+    {
+        safeUnsub(
+            seq_, [&] { source_.unsubAccountInternal(seq_, normalSubscriptions_, false); }, j);
+    }

    for (auto const& account : accountHistorySubscriptions_)
-        source_.unsubAccountHistoryInternal(seq_, account, false);
+    {
+        safeUnsub(seq_, [&] { source_.unsubAccountHistoryInternal(seq_, account, false); }, j);
+    }
+
+    for (auto const& book : bookSubscriptions_)
+    {
+        safeUnsub(seq_, [&] { source_.unsubBookInternal(seq_, book); }, j);
+    }
 }

 Resource::Consumer&
@@ -114,6 +165,20 @@ InfoSub::deleteSubAccountHistory(AccountID const& account)
    accountHistorySubscriptions_.erase(account);
 }

+void
+InfoSub::insertBookSubscription(Book const& book)
+{
+    std::scoped_lock const sl(lock_);
+    bookSubscriptions_.insert(book);
+}
+
+void
+InfoSub::deleteBookSubscription(Book const& book)
+{
+    std::scoped_lock const sl(lock_);
+    bookSubscriptions_.erase(book);
+}
+
 void
 InfoSub::clearRequest()
 {
--- a/src/libxrpl/tx/transactors/payment_channel/PaymentChannelClaim.cpp
+++ b/src/libxrpl/tx/transactors/payment_channel/PaymentChannelClaim.cpp
@@ -42,6 +42,9 @@ PaymentChannelClaim::getFlagsMask(PreflightContext const&)
 NotTEC
 PaymentChannelClaim::preflight(PreflightContext const& ctx)
 {
+    if (ctx.rules.enabled(fixCleanup3_2_0) && ctx.tx[sfChannel] == beast::kZero)
+        return temMALFORMED;
+
    auto const bal = ctx.tx[~sfBalance];
    if (bal && (!isXRP(*bal) || *bal <= beast::kZero))
        return temBAD_AMOUNT;
@@ -116,12 +119,10 @@ PaymentChannelClaim::doApply()
    AccountID const txAccount = ctx_.tx[sfAccount];

    auto const curExpiration = (*slep)[~sfExpiration];
+    if (isChannelExpired(ctx_.view(), (*slep)[~sfCancelAfter]) ||
+        isChannelExpired(ctx_.view(), curExpiration))
    {
-        auto const cancelAfter = (*slep)[~sfCancelAfter];
-        auto const closeTime = ctx_.view().header().parentCloseTime.time_since_epoch().count();
-        if ((cancelAfter && closeTime >= *cancelAfter) ||
-            (curExpiration && closeTime >= *curExpiration))
-            return closeChannel(slep, ctx_.view(), k.key, ctx_.registry.get().getJournal("View"));
+        return closeChannel(slep, ctx_.view(), k.key, ctx_.registry.get().getJournal("View"));
    }

    if (txAccount != src && txAccount != dst)
@@ -134,13 +135,19 @@ PaymentChannelClaim::doApply()
        auto const reqBalance = ctx_.tx[sfBalance].xrp();

        if (txAccount == dst && !ctx_.tx[~sfSignature])
-            return temBAD_SIGNATURE;
+        {
+            return ctx_.view().rules().enabled(fixCleanup3_2_0) ? TER{tecNO_PERMISSION}
+                                                                : TER{temBAD_SIGNATURE};
+        }

        if (ctx_.tx[~sfSignature])
        {
            PublicKey const pk((*slep)[sfPublicKey]);
            if (ctx_.tx[sfPublicKey] != pk)
-                return temBAD_SIGNER;
+            {
+                return ctx_.view().rules().enabled(fixCleanup3_2_0) ? TER{tecNO_PERMISSION}
+                                                                    : TER{temBAD_SIGNER};
+            }
        }

        if (reqBalance > chanFunds)
@@ -184,9 +191,10 @@ PaymentChannelClaim::doApply()
        if (dst == txAccount || (*slep)[sfBalance] == (*slep)[sfAmount])
            return closeChannel(slep, ctx_.view(), k.key, ctx_.registry.get().getJournal("View"));

-        auto const settleExpiration =
-            ctx_.view().header().parentCloseTime.time_since_epoch().count() +
-            (*slep)[sfSettleDelay];
+        auto const settleExpiration = saturatingAdd(
+            ctx_.view().rules(),
+            ctx_.view().header().parentCloseTime.time_since_epoch().count(),
+            (*slep)[sfSettleDelay]);

        if (!curExpiration || *curExpiration > settleExpiration)
        {
--- a/src/libxrpl/tx/transactors/payment_channel/PaymentChannelFund.cpp
+++ b/src/libxrpl/tx/transactors/payment_channel/PaymentChannelFund.cpp
@@ -6,6 +6,7 @@
 #include <xrpl/ledger/ReadView.h>
 #include <xrpl/ledger/helpers/PaymentChannelHelpers.h>
 #include <xrpl/protocol/AccountID.h>
+#include <xrpl/protocol/Feature.h>
 #include <xrpl/protocol/Indexes.h>
 #include <xrpl/protocol/Keylet.h>
 #include <xrpl/protocol/LedgerFormats.h>
@@ -29,6 +30,9 @@ PaymentChannelFund::makeTxConsequences(PreflightContext const& ctx)
 NotTEC
 PaymentChannelFund::preflight(PreflightContext const& ctx)
 {
+    if (ctx.rules.enabled(fixCleanup3_2_0) && ctx.tx[sfChannel] == beast::kZero)
+        return temMALFORMED;
+
    if (!isXRP(ctx.tx[sfAmount]) || (ctx.tx[sfAmount] <= beast::kZero))
        return temBAD_AMOUNT;

@@ -45,13 +49,12 @@ PaymentChannelFund::doApply()

    AccountID const src = (*slep)[sfAccount];
    auto const txAccount = ctx_.tx[sfAccount];
-    auto const expiration = (*slep)[~sfExpiration];
+    auto const curExpiration = (*slep)[~sfExpiration];

+    if (isChannelExpired(ctx_.view(), (*slep)[~sfCancelAfter]) ||
+        isChannelExpired(ctx_.view(), curExpiration))
    {
-        auto const cancelAfter = (*slep)[~sfCancelAfter];
-        auto const closeTime = ctx_.view().header().parentCloseTime.time_since_epoch().count();
-        if ((cancelAfter && closeTime >= *cancelAfter) || (expiration && closeTime >= *expiration))
-            return closeChannel(slep, ctx_.view(), k.key, ctx_.registry.get().getJournal("View"));
+        return closeChannel(slep, ctx_.view(), k.key, ctx_.registry.get().getJournal("View"));
    }

    if (src != txAccount)
@@ -60,16 +63,21 @@ PaymentChannelFund::doApply()
        return tecNO_PERMISSION;
    }

-    if (auto extend = ctx_.tx[~sfExpiration])
+    if (auto newExpiration = ctx_.tx[~sfExpiration])
    {
-        auto minExpiration = ctx_.view().header().parentCloseTime.time_since_epoch().count() +
-            (*slep)[sfSettleDelay];
-        if (expiration && *expiration < minExpiration)
-            minExpiration = *expiration;
+        auto minExpiration = saturatingAdd(
+            ctx_.view().rules(),
+            ctx_.view().header().parentCloseTime.time_since_epoch().count(),
+            (*slep)[sfSettleDelay]);
+        if (curExpiration && *curExpiration < minExpiration)
+            minExpiration = *curExpiration;

-        if (*extend < minExpiration)
-            return temBAD_EXPIRATION;
-        (*slep)[~sfExpiration] = *extend;
+        if (*newExpiration < minExpiration)
+        {
+            return ctx_.view().rules().enabled(fixCleanup3_2_0) ? TER{tecNO_PERMISSION}
+                                                                : TER{temBAD_EXPIRATION};
+        }
+        (*slep)[~sfExpiration] = *newExpiration;
        ctx_.view().update(slep);
    }

--- a/src/test/app/PayChan_test.cpp
+++ b/src/test/app/PayChan_test.cpp
@@ -1990,7 +1990,10 @@ public:
    run() override
    {
        using namespace test::jtx;
-        FeatureBitset const all{testableAmendments()};
+        // fixCleanup3_2_0 changes payment-channel error codes (tem* -> tec*)
+        // and channel-closing semantics. This suite asserts the
+        // pre-amendment behavior, so run it with the amendment disabled.
+        FeatureBitset const all{testableAmendments() - fixCleanup3_2_0};
        testWithFeats(all);
        testDepositAuthCreds();
        testMetaAndOwnership(all - fixIncludeKeyletFields);
--- a/src/test/nodestore/import_test.cpp
+++ b/src/test/nodestore/import_test.cpp
@@ -21,11 +21,16 @@
 #include <nudb/file.hpp>
 #include <nudb/native_file.hpp>
 #include <nudb/xxhasher.hpp>
+
+#if XRPL_ROCKSDB_AVAILABLE
+
 #include <rocksdb/db.h>
 #include <rocksdb/iterator.h>
 #include <rocksdb/options.h>
 #include <rocksdb/status.h>

+#endif
+
 #include <algorithm>
 #include <chrono>
 #include <cmath>
--- a/src/test/overlay/TMGetObjectByHash_test.cpp
+++ b/src/test/overlay/TMGetObjectByHash_test.cpp
@@ -100,6 +100,17 @@ class TMGetObjectByHash_test : public beast::unit_test::Suite
            return lastSentMessage_;
        }

+        // Synchronous test access to the JobQueue-dispatched processor.
+        // The production path runs this on JtLedgerReq; tests need a
+        // synchronous entry point to inspect the reply via send().
+        // PeerImp::processGetObjectByHash is `protected` so the derived
+        // test subclass can call it directly.
+        void
+        runProcessGetObjectByHash(std::shared_ptr<protocol::TMGetObjectByHash> const& m)
+        {
+            processGetObjectByHash(m);
+        }
+
        static void
        resetId()
        {
@@ -179,6 +190,10 @@ class TMGetObjectByHash_test : public beast::unit_test::Suite
    /**
     * Test that reply is limited to hardMaxReplyNodes when more objects
     * are requested than the limit allows.
+     *
+     * `onMessage(TMGetObjectByHash)` dispatches the generic-query path
+     * to the JobQueue, so tests invoke the synchronous processor
+     * directly via `runProcessGetObjectByHash`.
     */
    void
    testReplyLimit(size_t const numObjects, int const expectedReplySize)
@@ -191,8 +206,7 @@ class TMGetObjectByHash_test : public beast::unit_test::Suite
        auto peer = createPeer(env);

        auto request = createRequest(numObjects, env);
-        // Call the onMessage handler
-        peer->onMessage(request);
+        peer->runProcessGetObjectByHash(request);

        // Verify that a reply was sent
        auto sentMessage = peer->getLastSentMessage();
--- a/src/xrpld/app/ledger/OrderBookDBImpl.cpp
+++ b/src/xrpld/app/ledger/OrderBookDBImpl.cpp
@@ -9,19 +9,17 @@
 #include <xrpl/core/JobQueue.h>
 #include <xrpl/core/ServiceRegistry.h>
 #include <xrpl/ledger/AcceptedLedgerTx.h>
-#include <xrpl/ledger/BookListeners.h>
 #include <xrpl/ledger/OrderBookDB.h>
+#include <xrpl/ledger/ReadView.h>
 #include <xrpl/protocol/Asset.h>
 #include <xrpl/protocol/Book.h>
 #include <xrpl/protocol/Issue.h>
 #include <xrpl/protocol/LedgerFormats.h>
-#include <xrpl/protocol/MultiApiJson.h>
 #include <xrpl/protocol/SField.h>
 #include <xrpl/protocol/UintTypes.h>
 #include <xrpl/server/NetworkOPs.h>
 #include <xrpl/shamap/SHAMapMissingNode.h>

-#include <cstdint>
 #include <exception>
 #include <memory>
 #include <mutex>
@@ -307,55 +305,10 @@ OrderBookDBImpl::isBookToXRP(Asset const& asset, std::optional<Domain> const& do
    return xrpBooks_.contains(asset);
 }

-BookListeners::pointer
-OrderBookDBImpl::makeBookListeners(Book const& book)
+hash_set<Book>
+affectedBooks(AcceptedLedgerTx const& alTx, beast::Journal const& j)
 {
-    std::scoped_lock const sl(lock_);
-    auto ret = getBookListeners(book);
-
-    if (!ret)
-    {
-        ret = std::make_shared<BookListeners>();
-
-        listeners_[book] = ret;
-        XRPL_ASSERT(
-            getBookListeners(book) == ret,
-            "xrpl::OrderBookDB::makeBookListeners : result roundtrip "
-            "lookup");
-    }
-
-    return ret;
-}
-
-BookListeners::pointer
-OrderBookDBImpl::getBookListeners(Book const& book)
-{
-    BookListeners::pointer ret;
-    std::scoped_lock const sl(lock_);
-
-    auto it0 = listeners_.find(book);
-    if (it0 != listeners_.end())
-        ret = it0->second;
-
-    return ret;
-}
-
-// Based on the meta, send the meta to the streams that are listening.
-// We need to determine which streams a given meta effects.
-void
-OrderBookDBImpl::processTxn(
-    std::shared_ptr<ReadView const> const& ledger,
-    AcceptedLedgerTx const& alTx,
-    MultiApiJson const& jvObj)
-{
-    std::scoped_lock const sl(lock_);
-
-    // For this particular transaction, maintain the set of unique
-    // subscriptions that have already published it.  This prevents sending
-    // the transaction multiple times if it touches multiple ltOFFER
-    // entries for the same book, or if it touches multiple books and a
-    // single client has subscribed to those books.
-    hash_set<std::uint64_t> havePublished;
+    hash_set<Book> result;

    for (auto const& node : alTx.getMeta().getNodes())
    {
@@ -363,40 +316,41 @@ OrderBookDBImpl::processTxn(
        {
            if (node.getFieldU16(sfLedgerEntryType) == ltOFFER)
            {
-                auto process = [&, this](SField const& field) {
+                auto extract = [&](SField const& field) {
                    if (auto data = dynamic_cast<STObject const*>(node.peekAtPField(field)); data &&
                        data->isFieldPresent(sfTakerPays) && data->isFieldPresent(sfTakerGets))
                    {
-                        auto listeners = getBookListeners(
-                            {data->getFieldAmount(sfTakerGets).asset(),
-                             data->getFieldAmount(sfTakerPays).asset(),
-                             (*data)[~sfDomainID]});
-                        if (listeners)
-                            listeners->publish(jvObj, havePublished);
+                        result.emplace(
+                            data->getFieldAmount(sfTakerGets).asset(),
+                            data->getFieldAmount(sfTakerPays).asset(),
+                            (*data)[~sfDomainID]);
                    }
                };

-                // We need a field that contains the TakerGets and TakerPays
-                // parameters.
                if (node.getFName() == sfModifiedNode)
                {
-                    process(sfPreviousFields);
+                    extract(sfPreviousFields);
                }
                else if (node.getFName() == sfCreatedNode)
                {
-                    process(sfNewFields);
+                    extract(sfNewFields);
                }
                else if (node.getFName() == sfDeletedNode)
                {
-                    process(sfFinalFields);
+                    extract(sfFinalFields);
                }
            }
        }
        catch (std::exception const& ex)
        {
-            JLOG(j_.info()) << "processTxn: field not found (" << ex.what() << ")";
+            // The bad node is skipped; other affected books in the same
+            // transaction are still returned. Logged at warn so a malformed
+            // offer node is visible to operators.
+            JLOG(j.warn()) << "affectedBooks: skipping malformed node (" << ex.what() << ")";
        }
    }
+
+    return result;
 }

 }  // namespace xrpl
--- a/src/xrpld/app/ledger/OrderBookDBImpl.h
+++ b/src/xrpld/app/ledger/OrderBookDBImpl.h
@@ -1,10 +1,7 @@
 #pragma once

 #include <xrpl/core/ServiceRegistry.h>
-#include <xrpl/ledger/AcceptedLedgerTx.h>
-#include <xrpl/ledger/BookListeners.h>
 #include <xrpl/ledger/OrderBookDB.h>
-#include <xrpl/protocol/MultiApiJson.h>
 #include <xrpl/protocol/UintTypes.h>

 #include <mutex>
@@ -54,18 +51,6 @@ public:
    void
    update(std::shared_ptr<ReadView const> const& ledger);

-    // see if this txn effects any orderbook
-    void
-    processTxn(
-        std::shared_ptr<ReadView const> const& ledger,
-        AcceptedLedgerTx const& alTx,
-        MultiApiJson const& jvObj) override;
-
-    BookListeners::pointer
-    getBookListeners(Book const&) override;
-    BookListeners::pointer
-    makeBookListeners(Book const&) override;
-
 private:
    std::reference_wrapper<ServiceRegistry> registry_;
    int const pathSearchMax_;
@@ -84,10 +69,6 @@ private:

    std::recursive_mutex lock_;

-    using BookToListenersMap = hash_map<Book, BookListeners::pointer>;
-
-    BookToListenersMap listeners_;
-
    std::atomic<std::uint32_t> seq_;

    beast::Journal const j_;
--- a/src/xrpld/app/misc/NetworkOPs.cpp
+++ b/src/xrpld/app/misc/NetworkOPs.cpp
@@ -527,6 +527,8 @@ public:
    updateLocalTx(ReadView const& view) override;
    std::size_t
    getLocalTxCount() override;
+    std::size_t
+    getBookSubscribersCount() override;

    //
    // Monitoring: publisher side.
@@ -586,7 +588,9 @@ public:
    bool
    subBook(InfoSub::ref ispListener, Book const&) override;
    bool
-    unsubBook(std::uint64_t uListener, Book const&) override;
+    unsubBook(InfoSub::ref ispListener, Book const&) override;
+    bool
+    unsubBookInternal(std::uint64_t uListener, Book const&) override;

    bool
    subManifests(InfoSub::ref ispListener) override;
@@ -629,6 +633,12 @@ public:
    bool
    tryRemoveRpcSub(std::string const& strUrl) override;

+    beast::Journal const&
+    journal() const override
+    {
+        return journal_;
+    }
+
    void
    stop() override
    {
@@ -705,6 +715,32 @@ private:
        AcceptedLedgerTx const& transaction,
        bool last);

+    /**
+     * Fan transaction notifications out to all book subscribers.
+     *
+     * Extracts the set of order books affected by @p transaction, then
+     * delivers @p jvObj to every live subscriber of those books.
+     *
+     * Uses a two-pass design to keep subLock_ hold time short:
+     *   1. Under subLock_, collect strong InfoSub pointers for all live
+     *      subscribers and prune any expired weak_ptrs encountered.
+     *   2. Release subLock_, then call send() on each collected pointer.
+     *
+     * @param transaction The accepted ledger transaction to inspect.
+     * @param jvObj JSON representation of the transaction to deliver.
+     *
+     * @note Thread-safety: acquires subLock_ for the collection pass only.
+     *       send() is intentionally called outside the lock to avoid blocking
+     *       all other sub/unsub/publish paths while I/O is in progress.
+     * @note Contention: subLock_ is shared with all other subscription types.
+     *       On high-throughput nodes processing multi-hop payments that touch
+     *       many offer nodes, this pass holds subLock_ longer than the old
+     *       per-book BookListeners locks did. This is an accepted trade-off
+     *       for lock-domain simplicity.
+     */
+    void
+    pubBookTransaction(AcceptedLedgerTx const& transaction, MultiApiJson const& jvObj);
+
    void
    pubProposedAccountTransaction(
        std::shared_ptr<ReadView const> const& ledger,
@@ -802,8 +838,19 @@ private:

    LedgerMaster& ledgerMaster_;

+    /** Maps each order book to its current set of subscribers.
+     *  Outer key: the Book (currency pair + optional domain).
+     *  Inner key: InfoSub::seq (unique per connection).
+     *  Inner value: weak_ptr so that a dropped connection does not prevent
+     *  the InfoSub from being destroyed; expired entries are pruned lazily
+     *  by pubBookTransaction and eagerly by unsubBookInternal (~InfoSub path).
+     *  Guarded by subLock_.
+     */
+    using SubBookMapType = hash_map<Book, SubMapType>;
+
    SubInfoMapType subAccount_;
    SubInfoMapType subRTAccount_;
+    SubBookMapType subBook_;  ///< Guarded by subLock_.

    subRpcMapType rpcSubMap_;

@@ -3192,6 +3239,16 @@ NetworkOPsImp::getLocalTxCount()
    return localTX_->size();
 }

+std::size_t
+NetworkOPsImp::getBookSubscribersCount()
+{
+    std::scoped_lock const sl(subLock_);
+    std::size_t total = 0;
+    for (auto const& [_, subs] : subBook_)
+        total += subs.size();
+    return total;
+}
+
 // This routine should only be used to publish accepted or validated
 // transactions.
 MultiApiJson
@@ -3353,11 +3410,89 @@ NetworkOPsImp::pubValidatedTransaction(
    }

    if (transaction.getResult() == tesSUCCESS)
-        registry_.get().getOrderBookDB().processTxn(ledger, transaction, jvObj);
+        pubBookTransaction(transaction, jvObj);

    pubAccountTransaction(ledger, transaction, last);
 }

+void
+NetworkOPsImp::pubBookTransaction(AcceptedLedgerTx const& alTx, MultiApiJson const& jvObj)
+{
+    auto const books = affectedBooks(alTx, journal_);
+    if (books.empty())
+        return;
+
+    // Two-pass design:
+    //
+    //   1. Under subLock_, walk subBook_, collect a strong pointer for each
+    //      unique listener (and prune any expired weak_ptrs we encounter).
+    //   2. Release subLock_, then send to each collected listener.
+    //
+    // Reasoning:
+    //   * send() can be slow / blocking, so holding subLock_ across it would
+    //     stall every other sub/unsub/pub path on this server (see the matching
+    //     TODO above pubServer at line ~2275).
+    //   * A strong pointer destructed while subLock_ is held risks running
+    //     ~InfoSub() in-line, which re-enters unsubBook() and mutates the very
+    //     subBook_/SubMapType being iterated -> dangling iterator UB.
+    //
+    // Releasing subLock_ before any InfoSub::pointer can decay solves both.
+    // ~InfoSub() reacquires subLock_ via unsubBook() on its own and serializes
+    // safely with concurrent traffic.
+
+    std::vector<InfoSub::pointer> listeners;
+    hash_set<std::uint64_t> seen;
+
+    // Sized for the common case where every affected book has at most
+    // one subscriber. Multi-subscriber books trigger reallocation, but
+    // that is rare and the upper-bound estimate (sum of per-book sizes)
+    // would itself require walking subBook_ twice.
+    listeners.reserve(books.size());
+    seen.reserve(books.size());
+
+    {
+        std::scoped_lock const sl(subLock_);
+
+        for (auto const& book : books)
+        {
+            auto it = subBook_.find(book);
+            if (it == subBook_.end())
+                continue;
+
+            for (auto sit = it->second.begin(); sit != it->second.end();)
+            {
+                if (auto p = sit->second.lock())
+                {
+                    // Defensive: subBook_ entries are normally cleared by
+                    // ~InfoSub() -> unsubBook(), so we rarely see expired
+                    // weak_ptrs here. The else branch covers the narrow race
+                    // where the last strong ref is dropped between insertion
+                    // and our lock() call.
+                    if (seen.emplace(p->getSeq()).second)
+                        listeners.emplace_back(std::move(p));
+                    ++sit;
+                }
+                else
+                {
+                    JLOG(journal_.debug())
+                        << "pubBookTransaction: pruning expired weak_ptr for seq=" << sit->first;
+                    sit = it->second.erase(sit);
+                }
+            }
+
+            if (it->second.empty())
+                subBook_.erase(it);
+        }
+    }
+
+    for (auto const& p : listeners)
+    {
+        jvObj.visit(p->getApiVersion(), [&](json::Value const& jv) { p->send(jv, true); });
+    }
+    // listeners destructs here, outside subLock_; ~InfoSub (if any fires)
+    // will reacquire subLock_ via unsubBook with no iterator hazard.
+}
+
 void
 NetworkOPsImp::pubAccountTransaction(
    std::shared_ptr<ReadView const> const& ledger,
@@ -4011,26 +4146,39 @@ NetworkOPsImp::unsubAccountHistoryInternal(
 bool
 NetworkOPsImp::subBook(InfoSub::ref isrListener, Book const& book)
 {
-    if (auto listeners = registry_.get().getOrderBookDB().makeBookListeners(book))
+    // Server-side insert first, then InfoSub bookkeeping. If the InfoSub-side
+    // insert throws, the orphan in subBook_ is cleared by the expired-weak_ptr
+    // prune in pubBookTransaction. With the reverse ordering, ~InfoSub would
+    // call unsubBookInternal for a key that was never inserted server-side.
    {
-        listeners->addSubscriber(isrListener);
-    }
-    else
-    {
-        // LCOV_EXCL_START
-        UNREACHABLE("xrpl::NetworkOPsImp::subBook : null book listeners");
-        // LCOV_EXCL_STOP
+        std::scoped_lock const sl(subLock_);
+        subBook_[book].try_emplace(isrListener->getSeq(), isrListener);
    }
+    isrListener->insertBookSubscription(book);
    return true;
 }

 bool
-NetworkOPsImp::unsubBook(std::uint64_t uSeq, Book const& book)
+NetworkOPsImp::unsubBook(InfoSub::ref isrListener, Book const& book)
 {
-    if (auto listeners = registry_.get().getOrderBookDB().getBookListeners(book))
-        listeners->removeSubscriber(uSeq);
+    // Mirrors unsubAccount: clear the per-subscriber tracking set first so
+    // ~InfoSub does not re-issue an unsubBookInternal for a book the caller
+    // already removed, then erase the server-side entry.
+    isrListener->deleteBookSubscription(book);
+    return unsubBookInternal(isrListener->getSeq(), book);
+}

-    return true;
+bool
+NetworkOPsImp::unsubBookInternal(std::uint64_t uSeq, Book const& book)
+{
+    std::scoped_lock const sl(subLock_);
+    auto it = subBook_.find(book);
+    if (it == subBook_.end())
+        return false;
+    bool const erased = it->second.erase(uSeq) != 0u;
+    if (it->second.empty())
+        subBook_.erase(it);
+    return erased;
 }

 std::uint32_t
--- a/src/xrpld/overlay/detail/PeerImp.cpp
+++ b/src/xrpld/overlay/detail/PeerImp.cpp
@@ -22,6 +22,7 @@
 #include <xrpld/peerfinder/PeerfinderManager.h>
 #include <xrpld/peerfinder/Slot.h>

+#include <xrpl/basics/Blob.h>
 #include <xrpl/basics/Log.h>
 #include <xrpl/basics/SHAMapHash.h>
 #include <xrpl/basics/Slice.h>
@@ -81,6 +82,7 @@
 #include <xrpl.pb.h>

 #include <algorithm>
+#include <atomic>
 #include <chrono>
 #include <cstddef>
 #include <cstdint>
@@ -393,12 +395,22 @@ void
 PeerImp::charge(Resource::Charge const& fee, std::string const& context)
 {
    dispatch(strand_, [this, self = shared_from_this(), fee, context]() {
-        if (usage_.charge(fee, context) == Resource::Disposition::Drop &&
+        if ((usage_.charge(fee, context) == Resource::Disposition::Drop) &&
            usage_.disconnect(pJournal_))
        {
-            // Sever the connection.
-            overlay_.incPeerDisconnectCharges();
-            fail("charge: Resources");
+            // Idempotent: only the first worker to observe Drop counts the
+            // metric and posts fail().  Without the guard, several queued
+            // workers can all see Drop before fail() lands on the strand,
+            // overcounting peerDisconnectsCharges_ and posting duplicate
+            // shutdowns.  fail(std::string const&) self-posts to strand_
+            // when invoked off-strand.
+            bool expected = false;
+            if (chargeDisconnectFired_.compare_exchange_strong(
+                    expected, true, std::memory_order_acq_rel))
+            {
+                overlay_.incPeerDisconnectCharges();
+                fail("charge: Resources");
+            }
        }
    });
 }
@@ -2034,7 +2046,7 @@ PeerImp::checkTracking(std::uint32_t validationSeq)
 void
 PeerImp::checkTracking(std::uint32_t seq1, std::uint32_t seq2)
 {
-    int const diff = std::max(seq1, seq2) - std::min(seq1, seq2);
+    std::uint32_t const diff = std::max(seq1, seq2) - std::min(seq1, seq2);

    if (diff < Tuning::kConvergedLedgerLimit)
    {
@@ -2475,63 +2487,63 @@ PeerImp::onMessage(std::shared_ptr<protocol::TMGetObjectByHash> const& m)
            return;
        }

-        protocol::TMGetObjectByHash reply;
-
-        reply.set_query(false);
-
-        reply.set_type(packet.type());
-
        if (packet.has_ledgerhash())
        {
            if (!stringIsUInt256Sized(packet.ledgerhash()))
            {
-                fee_.update(Resource::kFeeMalformedRequest, "ledger hash");
+                JLOG(pJournal_.debug()) << "GetObj: malformed ledgerhash from peer " << id_;
+                fee_.update(Resource::kFeeMalformedRequest, "get object ledger hash");
                return;
            }
-
-            reply.set_ledgerhash(packet.ledgerhash());
        }
-
-        fee_.update(Resource::kFeeModerateBurdenPeer, " received a get object by hash request");
-
-        // This is a very minimal implementation
-        for (int i = 0; i < packet.objects_size(); ++i)
+        // Reject oversized requests before touching the NodeStore.
+        // The legitimate upper bound (InboundLedger::getNeededHashes())
+        // is 8 hashes; anything beyond kHardMaxReplyNodes is non-conforming.
+        if (packet.objects_size() > Tuning::kHardMaxReplyNodes)
        {
-            auto const& obj = packet.objects(i);
-            if (obj.has_hash() && stringIsUInt256Sized(obj.hash()))
-            {
-                uint256 const hash = uint256::fromRaw(obj.hash());
-                // VFALCO TODO Move this someplace more sensible so we dont
-                //             need to inject the NodeStore interfaces.
-                std::uint32_t const seq{obj.has_ledgerseq() ? obj.ledgerseq() : 0};
-                auto nodeObject{app_.getNodeStore().fetchNodeObject(hash, seq)};
-                if (nodeObject)
-                {
-                    protocol::TMIndexedObject& newObj = *reply.add_objects();
-                    newObj.set_hash(hash.begin(), hash.size());
-                    newObj.set_data(&nodeObject->getData().front(), nodeObject->getData().size());
-
-                    if (obj.has_nodeid())
-                        newObj.set_index(obj.nodeid());
-                    if (obj.has_ledgerseq())
-                        newObj.set_ledgerseq(obj.ledgerseq());
-
-                    // Check if by adding this object, reply has reached its
-                    // limit
-                    if (reply.objects_size() >= Tuning::kHardMaxReplyNodes)
-                    {
-                        fee_.update(
-                            Resource::kFeeModerateBurdenPeer,
-                            "Reply limit reached. Truncating reply.");
-                        break;
-                    }
-                }
-            }
+            JLOG(pJournal_.warn())
+                << "GetObj: oversized request from peer " << id_ << " (" << packet.objects_size()
+                << " > " << Tuning::kHardMaxReplyNodes << ")";
+            fee_.update(Resource::kFeeInvalidData, "oversized get object request");
+            return;
        }

-        JLOG(pJournal_.trace()) << "GetObj: " << reply.objects_size() << " of "
-                                << packet.objects_size();
-        send(std::make_shared<Message>(reply, protocol::mtGET_OBJECTS));
+        // Dispatch heavy synchronous NodeStore lookups off the peer's
+        // I/O strand and onto the bounded job queue, mirroring the pattern
+        // used by processLedgerRequest.
+        std::weak_ptr<PeerImp> const weak = shared_from_this();
+        bool const queued = app_.getJobQueue().addJob(JtLedgerReq, "RcvGetObjByHash", [weak, m]() {
+            auto peer = weak.lock();
+            if (!peer)
+                return;
+            try
+            {
+                peer->processGetObjectByHash(m);
+            }
+            catch (std::exception const& e)
+            {
+                // Surface backend failures (NodeStore I/O, allocation)
+                // back through the resource model so a misbehaving peer
+                // is still accountable rather than silently dropped.
+                JLOG(peer->pJournal_.warn()) << "GetObj: handler threw: " << e.what();
+                peer->charge(Resource::kFeeRequestNoReply, "get object handler exception");
+            }
+        });
+        if (!queued)
+        {
+            // The JobQueue is no longer accepting new work (typically
+            // because it is shutting down / has been joined).
+            JLOG(pJournal_.warn()) << "GetObj: job queue refused request from peer " << id_;
+            return;
+        }
+
+        // Admission-time charge: a peer that floods enqueues would
+        // otherwise be billed only the trivial onMessageEnd fee per
+        // message until the JobQueue catches up, re-creating an
+        // uncharged DoS window. Charge the base burden up-front (after
+        // a successful enqueue); the per-lookup differential is added
+        // in the worker.
+        fee_.update(Resource::kFeeModerateBurdenPeer, "received a get object by hash request");
    }
    else
    {
@@ -2587,6 +2599,69 @@ PeerImp::onMessage(std::shared_ptr<protocol::TMGetObjectByHash> const& m)
    }
 }

+void
+PeerImp::processGetObjectByHash(std::shared_ptr<protocol::TMGetObjectByHash> const& m)
+{
+    protocol::TMGetObjectByHash const& packet = *m;
+
+    protocol::TMGetObjectByHash reply;
+    reply.set_query(false);
+    reply.set_type(packet.type());
+
+    if (packet.has_ledgerhash())
+    {
+        reply.set_ledgerhash(packet.ledgerhash());
+    }
+
+    // Defense in depth: caller (onMessage) already validates cheap
+    // structural properties of the request before dispatching here:
+    //   - objects_size() <= kHardMaxReplyNodes (oversize gate)
+    //   - if has_ledgerhash() then ledgerhash is uint256-sized
+    // The iteration cap below mirrors the oversize gate so this method
+    // remains safe if invoked directly by tests or future callers, and
+    // a peer cannot drive unbounded NodeStore lookups by sending
+    // non-existent hashes.
+    int const requested = packet.objects_size();
+    int const iterLimit = std::min(requested, Tuning::kHardMaxReplyNodes);
+
+    for (int i = 0; i < iterLimit; ++i)
+    {
+        auto const& obj = packet.objects(i);
+        if (!obj.has_hash() || !stringIsUInt256Sized(obj.hash()))
+            continue;
+
+        uint256 const hash = uint256::fromRaw(obj.hash());
+        // VFALCO TODO Move this someplace more sensible so we don't
+        //             need to inject the NodeStore interfaces.
+        std::uint32_t const seq{obj.has_ledgerseq() ? obj.ledgerseq() : 0};
+        auto const nodeObject = app_.getNodeStore().fetchNodeObject(hash, seq);
+        if (!nodeObject)
+            continue;
+
+        protocol::TMIndexedObject& newObj = *reply.add_objects();
+        newObj.set_hash(hash.begin(), hash.size());
+        auto const& data = nodeObject->getData();
+        newObj.set_data(data.data(), data.size());
+        if (obj.has_nodeid())
+            newObj.set_index(obj.nodeid());
+        if (obj.has_ledgerseq())
+            newObj.set_ledgerseq(obj.ledgerseq());
+    }
+
+    // Apply work-proportional charge. `charge()` posts the disconnect
+    // step (if any) back to strand_, so it is safe to call from this
+    // JobQueue worker thread.
+    charge(
+        // We pass `requested` directly here, instead of actual lookups done. Which could be
+        // std::min(packet.objects_size(), static_cast<int>(Tuning::kHardMaxReplyNodes));
+        // Because we want to charge as per the request size, to discourage large requests.
+        computeGetObjectByHashFee(requested, reply.objects_size()),
+        "processed get object by hash request");
+
+    JLOG(pJournal_.trace()) << "GetObj: " << reply.objects_size() << " of " << requested;
+    send(std::make_shared<Message>(reply, protocol::mtGET_OBJECTS));
+}
+
 void
 PeerImp::onMessage(std::shared_ptr<protocol::TMHaveTransactions> const& m)
 {
@@ -3414,6 +3489,53 @@ PeerImp::processLedgerRequest(std::shared_ptr<protocol::TMGetLedger> const& m)
    send(std::make_shared<Message>(ledgerData, protocol::mtLEDGER_DATA));
 }

+// Differential pricing helper.  Returns only the *dynamic* component
+// of the per-message charge — the base `kFeeModerateBurdenPeer` is
+// applied at admission time in `onMessage(TMGetObjectByHash)` so a
+// high traffic client pays for the message regardless of when (or
+// whether) the worker runs.
+//
+// Dynamic charge model:
+//
+//   billable       = max(0, requested - kFreeObjectsPerRequest)
+//   missed         = max(0, requested - found)
+//   billableMisses = min(missed, billable)        // misses billed first
+//   billableHits   = billable - billableMisses
+//   sizeBand       = (requested >  kBandMediumMax) ? kCostBandLarge
+//                  : (requested >  kBandSmallMax)  ? kCostBandMedium
+//                                                  : kCostBandSmall
+//   dynamic        = billableHits   * kCostPerLookupHit
+//                  + billableMisses * kCostPerLookupMiss
+//                  + sizeBand
+//
+// Misses are billed first against the billable budget because a node store
+// seek dominates a cache hit and because invalid hashes are ~100% miss by construction.
+Resource::Charge
+PeerImp::computeGetObjectByHashFee(int const requested, int const found)
+{
+    int const billable = std::max(0, requested - static_cast<int>(Tuning::kFreeObjectsPerRequest));
+    // Clamp `missed` so a future caller passing found > requested cannot
+    // produce a negative value that flips the hits/misses split.
+    int const missed = std::max(0, requested - found);
+    int const billableMisses = std::min(missed, billable);
+    int const billableHits = billable - billableMisses;
+
+    int sizeBand = Tuning::kCostBandSmall;
+    if (requested > Tuning::kBandMediumMax)
+    {
+        sizeBand = Tuning::kCostBandLarge;
+    }
+    else if (requested > Tuning::kBandSmallMax)
+    {
+        sizeBand = Tuning::kCostBandMedium;
+    }
+
+    int const dynamic = (billableHits * Tuning::kCostPerLookupHit) +
+        (billableMisses * Tuning::kCostPerLookupMiss) + sizeBand;
+
+    return Resource::Charge(dynamic, "GetObject differential");
+}
+
 int
 PeerImp::getScore(bool haveItem) const
 {
--- a/src/xrpld/overlay/detail/PeerImp.h
+++ b/src/xrpld/overlay/detail/PeerImp.h
@@ -147,6 +147,12 @@ private:
    protocol::TMStatusChange lastStatus_;
    Resource::Consumer usage_;
    ChargeWithContext fee_;
+
+    // One-shot guard so concurrent JobQueue workers cannot double-count
+    // the per-connection peer-disconnect-by-charge metric (and cannot
+    // post duplicate fail() calls) when several queued requests cross
+    // kDropThreshold before the first fail() lands on the strand.
+    std::atomic<bool> chargeDisconnectFired_{false};
    std::shared_ptr<PeerFinder::Slot> const slot_;
    boost::beast::multi_buffer readBuffer_;
    http_request_type request_;
@@ -624,6 +630,67 @@ private:

    void
    processLedgerRequest(std::shared_ptr<protocol::TMGetLedger> const& m);
+
+protected:
+    // Kept `protected` so test subclasses (see
+    // TMGetObjectByHash_test) can drive the
+    // synchronous processor and the differential-pricing helper without
+    // routing through the JobQueue or going through `friend` plumbing.
+    // Production callers reach these members only via
+    // `onMessage(TMGetObjectByHash)` → JobQueue → `processGetObjectByHash`.
+
+    /** Process a generic-query TMGetObjectByHash message.
+
+        Dispatched from `onMessage(TMGetObjectByHash)` to the JobQueue
+        (`JtLedgerReq`) so synchronous NodeStore lookups do not block the
+        peer's I/O strand. Caps iteration at `Tuning::kHardMaxReplyNodes`
+        regardless of hit/miss outcome and applies differential pricing
+        via `computeGetObjectByHashFee()` after the fetch loop completes.
+
+        @param m The protocol message containing requested object hashes.
+     */
+    void
+    processGetObjectByHash(std::shared_ptr<protocol::TMGetObjectByHash> const& m);
+
+    /** Compute the per-message resource charge for a TMGetObjectByHash
+        request based on how much work was actually performed.
+
+        The charge has three components on top of the base
+        `Resource::kFeeModerateBurdenPeer`:
+          - per-hit lookup cost (cheap; usually served from cache)
+          - per-miss lookup cost (expensive node store seeks)
+          - request-size band surcharge (escalates abusive batch sizes)
+
+        The first `Tuning::kFreeObjectsPerRequest` objects are free so
+        that legitimate `InboundLedger::getNeededHashes()` traffic
+        (at most 8 objects) is unaffected.
+
+        @param requested Number of objects requested by the message. This
+                          value is used for request-size pricing and may
+                          exceed `Tuning::kHardMaxReplyNodes` when this
+                          helper is called directly, even though processing
+                          caps the iterations to `Tuning::kHardMaxReplyNodes`.
+        @param found     Number of objects successfully returned in the
+                         reply.
+        @return A `Resource::Charge` whose cost reflects the work performed.
+     */
+    static Resource::Charge
+    computeGetObjectByHashFee(int const requested, int const found);
+
+    /** Read-only accessor for the accumulated peer-message charge.
+
+        Exposed at `protected` scope so test subclasses can verify the
+        oversized-request rejection path (Layer 1) without invoking the
+        full JobQueue handler. Production callers should never read this back —
+        the value is consumed by `charge()`/`disconnect()` internally.
+
+        @return The current `Resource::Charge` accumulated on `fee_`.
+     */
+    Resource::Charge
+    currentFeeCharge() const
+    {
+        return fee_.fee;
+    }
 };

 //------------------------------------------------------------------------------
--- a/src/xrpld/overlay/detail/Tuning.h
+++ b/src/xrpld/overlay/detail/Tuning.h
@@ -1,14 +1,18 @@
 #pragma once
+#include <xrpl/shamap/SHAMapInnerNode.h>
+
+#include <cstddef>
+#include <cstdint>

 namespace xrpl::Tuning {

 /** How many ledgers off a server can be and we will
    still consider it converged */
-static constexpr auto kConvergedLedgerLimit = 24;
+static constexpr std::uint32_t kConvergedLedgerLimit = 24;

 /** How many ledgers off a server has to be before we
    consider it diverged */
-static constexpr auto kDivergedLedgerLimit = 128;
+static constexpr std::uint32_t kDivergedLedgerLimit = 128;

 /** The soft cap on the number of ledger entries in a single reply. */
 static constexpr auto kSoftMaxReplyNodes = 8192;
@@ -37,4 +41,92 @@ static constexpr auto kMaxQueryDepth = 3;
 /** Size of buffer used to read from the socket. */
 constexpr std::size_t kReadBufferBytes = 16384;

+/** TMGetObjectByHash differential pricing.
+
+    Honest peers ask for at most 8 hashes per call (the header, or up to
+    4 state + 4 tx hashes from `InboundLedger::getNeededHashes()`). The
+    free tier covers them at zero cost. Beyond that, each lookup is billed:
+    'misses' cost much more than 'hits' because a miss does a node store seek
+    while a hit is usually served from cache. On top of that, a size-band
+    surcharge kicks in for larger requests so an attacker who crams a
+    single message with thousands of hashes blows past
+    `Resource::kDropThreshold` and gets disconnected.
+
+    The numbers below are picked to keep three things true given
+    `kDropThreshold = 25000`:
+
+      - Honest traffic (<= 8 objects per request) is free.
+      - A single all-miss request at `kHardMaxReplyNodes` (12288) costs
+        more than the drop threshold, so an attacker gets dropped in one
+        message.
+      - A peer spamming 1024-object hit-only requests gets dropped in
+        ~19 messages — fast enough to be useful, slow enough that an
+        honest peer momentarily sending oversized requests has time to
+        back off. */
+
+/** How many objects a request can ask for before per-lookup billing
+    begins?
+    Twice the honest peak (8) so a peer that occasionally retries a hash
+    never trips pricing. Same value as `SHAMapInnerNode::kBranchFactor`;
+    that's a coincidence, not a requirement. */
+static constexpr auto kFreeObjectsPerRequest = 16;
+
+/** Cost of one cache-hit lookup. The unit; everything else is a
+    multiple of this. */
+static constexpr auto kCostPerLookupHit = 1;
+
+/** Cost of one node-store miss, in units of `kCostPerLookupHit`.
+
+    A miss does a node store disk seek; a hit usually comes from cache.
+    The 8x ratio is an order-of-magnitude guess at the latency gap on
+    SSD-backed nodes, not a measured number. The math only requires this
+    to be at least 2 — any smaller and a full-miss request at the hard
+    cap wouldn't trip the drop threshold. 8 leaves headroom: if
+    `kDropThreshold` goes up or `kHardMaxReplyNodes` comes down, the
+    drop-on-attack property still holds without a code change. */
+static constexpr auto kCostPerLookupMiss = 8;
+
+/** Size-band surcharges. Whichever band a request's size falls into,
+    its surcharge is added once on top of the per-lookup cost.
+
+    The job of the surcharge is to make crossing a band edge feel like
+    a step, not a slope. With these values, the cost roughly doubles or triples at each cliff:
+
+        n=64: costs 48 => n=65 costs 149 (~3x jump)
+        n=1024: costs 1108 => n=1025 costs 2009 (~2x jump)
+
+    The 10x step between medium and large mirrors the ~16x step
+    between the band edges (64 -> 1024) so the cliff feels comparable
+    at both scales.
+ */
+static constexpr auto kCostBandSmall = 0;
+static constexpr auto kCostBandMedium = 100;
+static constexpr auto kCostBandLarge = 1000;
+
+/** How many hashes per type an honest peer asks for at a time.
+
+    Matches the `4` passed to `neededStateHashes(4)` and
+    `neededTxHashes(4)` in `InboundLedger::getNeededHashes()`. Kept here
+    instead of imported from the ledger module so overlay stays
+    self-contained; if that `4` ever changes, update this in lockstep or
+    the band thresholds below will start charging honest peers. */
+static constexpr auto kLegitHashesPerType = 4;
+
+/** Cutoffs that decide which size band a request falls into.
+
+    A SHAMap inner node has 16 children; an honest peer asks for 4
+    hashes per type. So:
+
+      kBandSmallMax  = 4 * 16    =   64  // one inner node's worth
+      kBandMediumMax = 4 * 16^2  = 1024  // a depth-2 subtree's worth
+
+    A request up to 64 objects is small (no surcharge); up to 1024 is
+    medium; anything larger is large. The bounds are inclusive: a
+    request of exactly 64 is small, 65 is medium. Anything past 1024 is
+    well beyond what the honest sync path produces, so it's billed at
+    the large rate to drive attack-shaped traffic over the drop
+    threshold quickly. */
+static constexpr auto kBandSmallMax = kLegitHashesPerType * SHAMapInnerNode::kBranchFactor;
+static constexpr auto kBandMediumMax = kBandSmallMax * SHAMapInnerNode::kBranchFactor;
+
 }  // namespace xrpl::Tuning
--- a/src/xrpld/rpc/handlers/subscribe/Unsubscribe.cpp
+++ b/src/xrpld/rpc/handlers/subscribe/Unsubscribe.cpp
@@ -186,13 +186,23 @@ doUnsubscribe(RPC::JsonContext& context)
                book.domain = domain;
            }

-            context.netOps.unsubBook(ispSub->getSeq(), book);
+            if (!context.netOps.unsubBook(ispSub, book))
+            {
+                JLOG(context.j.debug())
+                    << "doUnsubscribe: book not subscribed (no-op for seq=" << ispSub->getSeq()
+                    << ")";
+            }

            // both_sides is deprecated.
            if ((jv.isMember(jss::both) && jv[jss::both].asBool()) ||
                (jv.isMember(jss::both_sides) && jv[jss::both_sides].asBool()))
            {
-                context.netOps.unsubBook(ispSub->getSeq(), reversed(book));
+                if (!context.netOps.unsubBook(ispSub, reversed(book)))
+                {
+                    JLOG(context.j.debug())
+                        << "doUnsubscribe: reversed book not subscribed (no-op for seq="
+                        << ispSub->getSeq() << ")";
+                }
            }
        }
    }