Merge remote-tracking branch 'origin/dev' into external-env-tests-wd

# Conflicts:
#	src/test/app/AMMClawback_test.cpp
#	src/test/app/AMM_test.cpp
#	src/test/jtx/AMMTest.h
#	src/test/rpc/AMMInfo_test.cpp
#	src/xrpld/app/hook/applyHook.h
#	src/xrpld/app/hook/detail/applyHook.cpp
#	src/xrpld/app/tx/detail/AMMClawback.cpp
#	src/xrpld/app/tx/detail/InvariantCheck.cpp
#	src/xrpld/app/tx/detail/InvariantCheck.h

Builds/levelization/results/ordering.txt

@@ -77,11 +77,6 @@ test.ledger > xrpld.app
 test.ledger > xrpld.core
 test.ledger > xrpld.ledger
 test.ledger > xrpl.protocol
-test.net > test.toplevel
-test.net > xrpl.basics
-test.net > xrpld.core
-test.net > xrpld.net
-test.net > xrpl.json
 test.nodestore > test.jtx
 test.nodestore > test.toplevel
 test.nodestore > test.unit_test

build-core.sh

@@ -95,8 +95,16 @@ if [[ "$4" == "" ]]; then
   echo "Non GH, local building, no Action runner magic"
 else
   # GH Action, runner
-  cp /io/release-build/xahaud /data/builds/$(date +%Y).$(date +%-m).$(date +%-d)-$(git rev-parse --abbrev-ref HEAD)+$4
-  cp /io/release-build/release.info /data/builds/$(date +%Y).$(date +%-m).$(date +%-d)-$(git rev-parse --abbrev-ref HEAD)+$4.releaseinfo
+    if [[ "$(git rev-parse --abbrev-ref HEAD)" == "release" ]]; then
+        echo "building on the release branch... placing it in builds/candidate"
+        mkdir /data/builds/candidate
+        cp /io/release-build/xahaud /data/builds/candidate/$(date +%Y).$(date +%-m).$(date +%-d)-$(git rev-parse --abbrev-ref HEAD)+$4
+        cp /io/release-build/release.info /data/builds/candidate/$(date +%Y).$(date +%-m).$(date +%-d)-$(git rev-parse --abbrev-ref HEAD)+$4.releaseinfo
+    else
+        echo "building non-release branch, placing it in builds root"
+        cp /io/release-build/xahaud /data/builds/$(date +%Y).$(date +%-m).$(date +%-d)-$(git rev-parse --abbrev-ref HEAD)+$4
+        cp /io/release-build/release.info /data/builds/$(date +%Y).$(date +%-m).$(date +%-d)-$(git rev-parse --abbrev-ref HEAD)+$4.releaseinfo
+    fi  
   echo "Published build to: http://build.xahau.tech/"
   echo $(date +%Y).$(date +%-m).$(date +%-d)-$(git rev-parse --abbrev-ref HEAD)+$4
 fi

cmake/RippledCore.cmake

@@ -179,7 +179,108 @@ if(xrpld)
     file(GLOB_RECURSE sources CONFIGURE_DEPENDS
       "${CMAKE_CURRENT_SOURCE_DIR}/src/test/*.cpp"
     )
+    if(HOOKS_TEST_ONLY OR DEFINED ENV{HOOKS_TEST_ONLY})
+      # Keep test infra but drop the individual *_test.cpp files
+      list(FILTER sources EXCLUDE REGEX "_test\\.cpp$")
+      message(STATUS "HOOKS_TEST_ONLY: excluded *_test.cpp from src/test/")
+    endif()
     target_sources(rippled PRIVATE ${sources})
+
+    # Optional: include external hook test sources from another directory.
+    # Set via -DHOOKS_TEST_DIR=/path/to/tests or env HOOKS_TEST_DIR.
+    # Optionally set HOOKS_C_DIR to pass --hooks-c-dir args to the compiler
+    # (e.g. "tipbot=/path/to/hooks" — multiple values separated by ";").
+    #
+    # hookz build-test-hooks must be on PATH. It auto-compiles hooks referenced
+    # in each *_test.cpp and generates *_test_hooks.h next to the test file.
+    if(NOT HOOKS_TEST_DIR AND DEFINED ENV{HOOKS_TEST_DIR})
+      set(HOOKS_TEST_DIR $ENV{HOOKS_TEST_DIR})
+    endif()
+    if(NOT HOOKS_C_DIR AND DEFINED ENV{HOOKS_C_DIR})
+      set(HOOKS_C_DIR $ENV{HOOKS_C_DIR})
+    endif()
+    if(HOOKS_TEST_DIR AND EXISTS "${HOOKS_TEST_DIR}")
+      file(GLOB EXTERNAL_HOOK_TESTS CONFIGURE_DEPENDS
+        "${HOOKS_TEST_DIR}/*_test.cpp"
+      )
+      if(EXTERNAL_HOOK_TESTS)
+        # Build extra args for hookz build-test-hooks
+        set(_hooks_extra_args "")
+        set(_hooks_source_deps "")
+        if(HOOKS_C_DIR)
+          foreach(_dir ${HOOKS_C_DIR})
+            list(APPEND _hooks_extra_args "--hooks-c-dir" "${_dir}")
+
+            string(REGEX REPLACE "^[^=]+=" "" _hook_dir "${_dir}")
+            if(EXISTS "${_hook_dir}")
+              file(GLOB_RECURSE _hook_dir_deps CONFIGURE_DEPENDS
+                "${_hook_dir}/*.c"
+                "${_hook_dir}/*.h"
+              )
+              if(HOOKS_TEST_DIR)
+                list(FILTER _hook_dir_deps EXCLUDE REGEX "^${HOOKS_TEST_DIR}/")
+              endif()
+              list(APPEND _hooks_source_deps ${_hook_dir_deps})
+            endif()
+          endforeach()
+          list(REMOVE_DUPLICATES _hooks_source_deps)
+        endif()
+        if(HOOKS_COVERAGE OR DEFINED ENV{HOOKS_COVERAGE})
+          list(APPEND _hooks_extra_args "--hook-coverage")
+          message(STATUS "Hook coverage enabled: compiling hooks with hookz")
+        endif()
+        if(HOOKS_FORCE_RECOMPILE OR DEFINED ENV{HOOKS_FORCE_RECOMPILE})
+          list(APPEND _hooks_extra_args "--force-write" "--no-cache")
+          message(STATUS "Hook force recompile enabled (cache bypassed)")
+        endif()
+
+        # Run hookz build-test-hooks on each test file before compilation
+        foreach(_test_file ${EXTERNAL_HOOK_TESTS})
+          get_filename_component(_stem ${_test_file} NAME_WE)
+          set(_hooks_header "${HOOKS_TEST_DIR}/${_stem}_hooks.h")
+          if(HOOKS_FORCE_RECOMPILE OR DEFINED ENV{HOOKS_FORCE_RECOMPILE})
+            # Always run — no DEPENDS, no OUTPUT caching
+            add_custom_target(compile_hooks_${_stem} ALL
+              COMMAND hookz build-test-hooks "${_test_file}" ${_hooks_extra_args}
+              WORKING_DIRECTORY "${CMAKE_SOURCE_DIR}"
+              COMMENT "Compiling hooks for ${_stem} (forced)"
+              VERBATIM
+            )
+            list(APPEND EXTERNAL_HOOK_TARGETS compile_hooks_${_stem})
+          else()
+            add_custom_command(
+              OUTPUT "${_hooks_header}"
+              COMMAND hookz build-test-hooks "${_test_file}" ${_hooks_extra_args}
+              DEPENDS "${_test_file}" ${_hooks_source_deps}
+              WORKING_DIRECTORY "${CMAKE_SOURCE_DIR}"
+              COMMENT "Compiling hooks for ${_stem}"
+              VERBATIM
+            )
+            list(APPEND EXTERNAL_HOOK_HEADERS "${_hooks_header}")
+          endif()
+        endforeach()
+
+        # Ensure headers are generated before rippled compiles
+        if(HOOKS_FORCE_RECOMPILE OR DEFINED ENV{HOOKS_FORCE_RECOMPILE})
+          foreach(_tgt ${EXTERNAL_HOOK_TARGETS})
+            add_dependencies(rippled ${_tgt})
+          endforeach()
+        else()
+          add_custom_target(compile_external_hooks DEPENDS ${EXTERNAL_HOOK_HEADERS})
+          add_dependencies(rippled compile_external_hooks)
+        endif()
+
+        target_sources(rippled PRIVATE ${EXTERNAL_HOOK_TESTS})
+        # Keep the generated hook-header include path scoped to the external
+        # test sources so changing HOOKS_TEST_DIR doesn't invalidate the
+        # compile command for the rest of rippled.
+        set_property(
+          SOURCE ${EXTERNAL_HOOK_TESTS}
+          APPEND PROPERTY INCLUDE_DIRECTORIES "${HOOKS_TEST_DIR}"
+        )
+        message(STATUS "Including external hook tests from: ${HOOKS_TEST_DIR}")
+      endif()
+    endif()
   endif()
 
   target_link_libraries(rippled

include/xrpl/basics/Log.h

@@ -27,6 +27,7 @@
 #include <fstream>
 #include <map>
 #include <memory>
+#include <functional>
 #include <mutex>
 #include <utility>
 
@@ -165,6 +166,7 @@ private:
     beast::severities::Severity thresh_;
     File file_;
     bool silent_ = false;
+    std::function<std::string(std::string const&)> transform_;
 
 public:
     Logs(beast::severities::Severity level);
@@ -203,6 +205,33 @@ public:
         std::string const& text,
         bool console);
 
+    /** Set a transform applied to every log message before output.
+     *  Useful in tests to replace raw account IDs with human-readable names.
+     *  Pass nullptr to clear.
+     *
+     *  TODO: This is test-only infrastructure (used by TestEnv). Consider
+     *  moving to SuiteLogs or a test-specific subclass if the Logs interface
+     *  needs to stay clean for production.
+     */
+    void
+    setTransform(std::function<std::string(std::string const&)> fn)
+    {
+        std::lock_guard lock(mutex_);
+        transform_ = std::move(fn);
+    }
+
+    /** Apply the current transform to text (or return as-is if none set). */
+    std::string const&
+    applyTransform(std::string const& text) const
+    {
+        if (!transform_)
+            return text;
+        // Store in thread_local to return a const ref
+        thread_local std::string buf;
+        buf = transform_(text);
+        return buf;
+    }
+
     std::string
     rotate();
 

include/xrpl/hook/Enum.h

@@ -417,6 +417,7 @@ getImportWhitelist(Rules const& rules)
 #define int64_t 0x7EU
 #define int32_t 0x7FU
 #define uint32_t 0x7FU
+#define void_t 0x00U
 
 #define HOOK_WRAP_PARAMS(...) __VA_ARGS__
 
@@ -428,11 +429,15 @@ getImportWhitelist(Rules const& rules)
 
 #include "hook_api.macro"
 
+    // Coverage callback: void __on_source_line(uint32_t line, uint32_t col)
+    whitelist["__on_source_line"] = {void_t, uint32_t, uint32_t};
+
 #undef HOOK_API_DEFINITION
 #undef HOOK_WRAP_PARAMS
 #undef int64_t
 #undef int32_t
 #undef uint32_t
+#undef void_t
 #pragma pop_macro("HOOK_API_DEFINITION")
 
     return whitelist;

include/xrpl/hook/Guard.h

@@ -1383,21 +1383,52 @@ validateGuards(
                 int result_count = parseLeb128(wasm, i, &i);
                 CHECK_SHORT_HOOK();
 
-                // this needs a reliable hook cleaner otherwise it will catch
-                // most compilers out
-                if (result_count != 1)
+                if (j == hook_type_idx)
                 {
-                    GUARDLOG(hook::log::FUNC_RETURN_COUNT)
-                        << "Malformed transaction. "
-                        << "Hook declares a function type that returns fewer "
-                           "or more than one value. "
-                        << "\n";
-                    return {};
+                    // hook/cbak must return exactly one value (i64)
+                    if (result_count != 1)
+                    {
+                        GUARDLOG(hook::log::FUNC_RETURN_COUNT)
+                            << "Malformed transaction. "
+                            << "hook/cbak function type must return exactly "
+                               "one value. "
+                            << "\n";
+                        return {};
+                    }
+                }
+                else if (first_signature)
+                {
+                    // For whitelisted imports, check expected return count.
+                    // void_t (0x00) means 0 return values.
+                    uint8_t expected_return =
+                        (*first_signature).get()[0];
+                    int expected_result_count =
+                        (expected_return == 0x00U) ? 0 : 1;
+                    if (result_count != expected_result_count)
+                    {
+                        GUARDLOG(hook::log::FUNC_RETURN_COUNT)
+                            << "Malformed transaction. "
+                            << "Hook API: " << *first_name
+                            << " has wrong return count "
+                            << "(expected " << expected_result_count
+                            << ", got " << result_count << ")."
+                            << "\n";
+                        return {};
+                    }
+                }
+                else
+                {
+                    if (result_count != 1)
+                    {
+                        GUARDLOG(hook::log::FUNC_RETURN_COUNT)
+                            << "Malformed transaction. "
+                            << "Hook declares a function type that returns "
+                               "fewer or more than one value. "
+                            << "\n";
+                        return {};
+                    }
                 }
 
-                // this can only ever be 1 in production, but in testing it may
-                // also be 0 or >1 so for completeness this loop is here but can
-                // be taken out in prod
                 for (int k = 0; k < result_count; ++k)
                 {
                     int result_type = parseLeb128(wasm, i, &i);

include/xrpl/hook/Macro.h

@@ -157,6 +157,7 @@
         [[maybe_unused]] ApplyContext& applyCtx = hookCtx.applyCtx;  \
         [[maybe_unused]] auto& view = applyCtx.view();               \
         [[maybe_unused]] auto j = applyCtx.app.journal("View");      \
+        [[maybe_unused]] auto jh = applyCtx.app.journal("HooksTrace"); \
         [[maybe_unused]] WasmEdge_MemoryInstanceContext* memoryCtx = \
             WasmEdge_CallingFrameGetMemoryInstance(&frameCtx, 0);    \
         [[maybe_unused]] unsigned char* memory =                     \

src/libxrpl/basics/Log.cpp

@@ -196,9 +196,10 @@ Logs::write(
     std::string const& text,
     bool console)
 {
-    std::string s;
-    format(s, text, level, partition);
     std::lock_guard lock(mutex_);
+    std::string const& transformed = transform_ ? transform_(text) : text;
+    std::string s;
+    format(s, transformed, level, partition);
     file_.writeln(s);
     if (!silent_)
         std::cerr << s << '\n';

src/test/jtx/Env.h

@@ -106,7 +106,8 @@ public:
         std::string const& partition,
         beast::severities::Severity threshold) override
     {
-        return std::make_unique<SuiteJournalSink>(partition, threshold, suite_);
+        return std::make_unique<SuiteJournalSink>(
+            partition, threshold, suite_, this);
     }
 };
 

src/test/jtx/TestEnv.h

@@ -0,0 +1,148 @@
+#ifndef TEST_JTX_TESTENV_H_INCLUDED
+#define TEST_JTX_TESTENV_H_INCLUDED
+
+#include <test/jtx/Env.h>
+#include <xrpl/basics/Log.h>
+#include <xrpl/protocol/AccountID.h>
+#include <cstdlib>
+#include <cstring>
+#include <map>
+#include <sstream>
+#include <string>
+
+namespace ripple {
+namespace test {
+namespace jtx {
+
+/**
+ * TestEnv wraps Env with:
+ *   - Named account registry: env.account("alice")
+ *   - Auto log transform: replaces r-addresses with Account(name) in log output
+ *   - Env-var driven per-partition log levels via TESTENV_LOGGING
+ *
+ * Usage:
+ *   TestEnv env{suite, features};
+ *   auto const& alice = env.account("alice");
+ *   auto const& bob = env.account("bob");
+ *   env.fund(XRP(10000), alice, bob);
+ *   // Logs now show Account(alice), Account(bob) instead of r-addresses
+ *
+ * Log levels via env var:
+ *   TESTENV_LOGGING="HooksTrace=trace,View=debug"
+ *
+ * Valid levels: trace, debug, info, warning, error, fatal
+ */
+class TestEnv : public Env
+{
+    std::map<std::string, Account> accounts_;
+    std::string prefix_;
+
+public:
+    TestEnv(beast::unit_test::suite& suite, FeatureBitset features)
+        : Env(suite, features)
+    {
+        installTransform();
+        applyLoggingEnvVar();
+    }
+
+    TestEnv(
+        beast::unit_test::suite& suite,
+        std::unique_ptr<Config> config,
+        FeatureBitset features,
+        std::unique_ptr<Logs> logs = nullptr,
+        beast::severities::Severity thresh = beast::severities::kError)
+        : Env(suite, std::move(config), features, std::move(logs), thresh)
+    {
+        installTransform();
+        applyLoggingEnvVar();
+    }
+
+    ~TestEnv()
+    {
+        app().logs().setTransform(nullptr);
+    }
+
+    /// Get or create a named account.
+    /// First call creates the Account; subsequent calls return the same one.
+    Account const&
+    account(std::string const& name)
+    {
+        auto [it, inserted] = accounts_.try_emplace(name, name);
+        return it->second;
+    }
+
+    /// Set a prefix that appears at the start of every log line.
+    /// Useful for visually separating test phases in trace output.
+    /// Pass empty string to clear.
+    void
+    setPrefix(std::string const& prefix)
+    {
+        prefix_ = prefix.empty() ? "" : "[" + prefix + "] ";
+    }
+
+private:
+    static beast::severities::Severity
+    parseSeverity(std::string const& s)
+    {
+        if (s == "trace")
+            return beast::severities::kTrace;
+        if (s == "debug")
+            return beast::severities::kDebug;
+        if (s == "info")
+            return beast::severities::kInfo;
+        if (s == "warning")
+            return beast::severities::kWarning;
+        if (s == "error")
+            return beast::severities::kError;
+        if (s == "fatal")
+            return beast::severities::kFatal;
+        return beast::severities::kError;
+    }
+
+    void
+    applyLoggingEnvVar()
+    {
+        // Parse TESTENV_LOGGING="Partition1=level,Partition2=level"
+        auto const* envVal = std::getenv("TESTENV_LOGGING");
+        if (!envVal || !envVal[0])
+            return;
+
+        std::istringstream ss(envVal);
+        std::string pair;
+        while (std::getline(ss, pair, ','))
+        {
+            auto eq = pair.find('=');
+            if (eq == std::string::npos)
+                continue;
+            auto partition = pair.substr(0, eq);
+            auto level = pair.substr(eq + 1);
+            app().logs().get(partition).threshold(parseSeverity(level));
+        }
+    }
+
+    void
+    installTransform()
+    {
+        app().logs().setTransform([this](std::string const& text) {
+            std::string out = prefix_ + text;
+            for (auto const& [name, acc] : accounts_)
+            {
+                auto raddr = toBase58(acc.id());
+                std::string::size_type pos = 0;
+                std::string replacement = "Account(" + name + ")";
+                while ((pos = out.find(raddr, pos)) != std::string::npos)
+                {
+                    out.replace(pos, raddr.size(), replacement);
+                    pos += replacement.size();
+                }
+            }
+            return out;
+        });
+    }
+};
+
+}  // namespace jtx
+}  // namespace test
+}  // namespace ripple
+
+#endif

src/test/net/RPCSub_test.cpp

@@ -1,351 +0,0 @@
-//------------------------------------------------------------------------------
-/*
-    This file is part of rippled: https://github.com/ripple/rippled
-    Copyright (c) 2024 Ripple Labs Inc.
-
-    Permission to use, copy, modify, and/or distribute this software for any
-    purpose  with  or without fee is hereby granted, provided that the above
-    copyright notice and this permission notice appear in all copies.
-
-    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
-    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
-    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
-    ANY  SPECIAL ,  DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
-    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
-    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
-    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
-*/
-//==============================================================================
-
-#include <test/jtx.h>
-#include <xrpld/core/Job.h>
-#include <xrpld/core/JobQueue.h>
-#include <xrpld/net/RPCSub.h>
-#include <xrpl/json/json_value.h>
-
-#include <boost/asio.hpp>
-#include <boost/asio/ip/tcp.hpp>
-
-#include <atomic>
-#include <chrono>
-#include <memory>
-#include <string>
-#include <thread>
-
-namespace ripple {
-namespace test {
-
-// Minimal HTTP endpoint that counts received webhook POSTs and replies
-// with a configurable status. Responses are EOF-delimited (no
-// Content-Length) and the socket is closed right after writing — the
-// exact shape that triggered the original handleData EOF-completion
-// leak. So these tests exercise RPCSub flow control AND the HTTPClient
-// EOF fix end to end: if either regressed, delivery would stall and the
-// expected count would never be reached within the timeout.
-class MockWebhookEndpoint
-{
-    boost::asio::io_service ios_;
-    std::unique_ptr<boost::asio::io_service::work> work_;
-    boost::asio::ip::tcp::acceptor acceptor_;
-    std::thread thread_;
-    unsigned short port_;
-
-    std::atomic<int> received_{0};
-    std::atomic<int> status_{200};
-    std::atomic<int> delayMs_{0};
-
-public:
-    MockWebhookEndpoint()
-        : work_(std::make_unique<boost::asio::io_service::work>(ios_))
-        , acceptor_(
-              ios_,
-              boost::asio::ip::tcp::endpoint(
-                  boost::asio::ip::address::from_string("127.0.0.1"),
-                  0))
-    {
-        port_ = acceptor_.local_endpoint().port();
-        accept();
-        thread_ = std::thread([this] { ios_.run(); });
-    }
-
-    ~MockWebhookEndpoint()
-    {
-        work_.reset();
-        boost::system::error_code ec;
-        acceptor_.close(ec);
-        ios_.stop();
-        if (thread_.joinable())
-            thread_.join();
-    }
-
-    unsigned short
-    port() const
-    {
-        return port_;
-    }
-
-    int
-    received() const
-    {
-        return received_;
-    }
-
-    void
-    setStatus(int s)
-    {
-        status_ = s;
-    }
-
-    // Delay each reply so delivery is deterministically slower than the
-    // microsecond-fast enqueue loop — keeps the deque full for the
-    // queue-cap drop test regardless of scheduling.
-    void
-    setResponseDelay(int ms)
-    {
-        delayMs_ = ms;
-    }
-
-private:
-    void
-    accept()
-    {
-        auto sock = std::make_shared<boost::asio::ip::tcp::socket>(ios_);
-        acceptor_.async_accept(*sock, [this, sock](auto ec) {
-            if (ec)
-                return;
-            handle(sock);
-            accept();
-        });
-    }
-
-    void
-    handle(std::shared_ptr<boost::asio::ip::tcp::socket> sock)
-    {
-        auto buf = std::make_shared<boost::asio::streambuf>();
-        boost::asio::async_read_until(
-            *sock, *buf, "\r\n\r\n", [this, sock, buf](auto ec, std::size_t) {
-                if (ec)
-                    return;
-
-                ++received_;
-
-                auto const delay = delayMs_.load();
-                if (delay > 0)
-                {
-                    auto timer =
-                        std::make_shared<boost::asio::steady_timer>(ios_);
-                    timer->expires_from_now(std::chrono::milliseconds(delay));
-                    timer->async_wait(
-                        [this, sock, timer](auto) { reply(sock); });
-                }
-                else
-                {
-                    reply(sock);
-                }
-            });
-    }
-
-    void
-    reply(std::shared_ptr<boost::asio::ip::tcp::socket> sock)
-    {
-        // EOF-delimited reply: no Content-Length, close after writing.
-        // This is the realistic failing-webhook shape.
-        auto resp = std::make_shared<std::string>(
-            "HTTP/1.0 " + std::to_string(status_.load()) +
-            " Reply\r\n\r\n{\"result\":{}}");
-        boost::asio::async_write(
-            *sock, boost::asio::buffer(*resp), [sock, resp](auto, std::size_t) {
-                boost::system::error_code ig;
-                sock->shutdown(boost::asio::ip::tcp::socket::shutdown_both, ig);
-                sock->close(ig);
-            });
-    }
-};
-
-//------------------------------------------------------------------------------
-
-class RPCSub_test : public beast::unit_test::suite
-{
-    // Generous ceiling: the instrumented Debug (coverage) build is much
-    // slower than Release, so timeouts are sized for that, not Release.
-    template <class Cond>
-    bool
-    waitFor(Cond cond, std::chrono::seconds timeout = std::chrono::seconds{30})
-    {
-        auto const deadline = std::chrono::steady_clock::now() + timeout;
-        while (!cond() && std::chrono::steady_clock::now() < deadline)
-            std::this_thread::sleep_for(std::chrono::milliseconds(10));
-        return cond();
-    }
-
-    std::shared_ptr<RPCSub>
-    makeSub(
-        jtx::Env& env,
-        MockWebhookEndpoint& ep,
-        std::size_t maxQueueSize = 16384)
-    {
-        return make_RPCSub(
-            env.app().getOPs(),
-            env.app().getJobQueue(),
-            "http://127.0.0.1:" + std::to_string(ep.port()) + "/",
-            "",
-            "",
-            env.app().logs(),
-            maxQueueSize);
-    }
-
-    // True once no RPCSub sending job is queued or running. sendThread
-    // captures a raw `this`, so the RPCSub must not be destroyed while a
-    // job is still in flight — wait on this before letting the sub die.
-    bool
-    sendingIdle(jtx::Env& env)
-    {
-        return env.app().getJobQueue().getJobCountTotal(jtCLIENT_SUBSCRIBE) ==
-            0;
-    }
-
-    // Wait for all events to reach the endpoint AND the sending job to
-    // finish, so the sub can be torn down without racing sendThread.
-    void
-    drainAndSettle(jtx::Env& env, MockWebhookEndpoint& ep, int expected)
-    {
-        bool const delivered =
-            waitFor([&] { return ep.received() >= expected; });
-        bool const idle = waitFor([&] { return sendingIdle(env); });
-        log << "  drainAndSettle: received=" << ep.received() << "/" << expected
-            << " idle=" << idle << std::endl;
-        BEAST_EXPECT(delivered);
-        BEAST_EXPECT(idle);
-    }
-
-    void
-    send(std::shared_ptr<RPCSub> const& sub, int n)
-    {
-        Json::Value ev(Json::objectValue);
-        ev["n"] = n;
-        sub->send(ev, false);
-    }
-
-    void
-    testDelivery()
-    {
-        testcase("Webhook events are delivered");
-
-        using namespace jtx;
-        Env env{*this};
-        MockWebhookEndpoint ep;
-
-        static constexpr int N = 10;
-        {
-            auto sub = makeSub(env, ep);
-            for (int i = 0; i < N; ++i)
-                send(sub, i);
-            drainAndSettle(env, ep, N);
-        }
-
-        BEAST_EXPECT(ep.received() == N);
-    }
-
-    void
-    testErrorsDoNotStall()
-    {
-        testcase("Delivery continues when endpoint returns HTTP 500");
-
-        // The original bug (xrpld #6341): an endpoint returning errors
-        // without Content-Length never completed, stalling delivery to
-        // ALL subscribers. Here every response is a 500 with no
-        // Content-Length (EOF-delimited) — all N must still arrive.
-        using namespace jtx;
-        Env env{*this};
-        MockWebhookEndpoint ep;
-        ep.setStatus(500);
-
-        static constexpr int N = 10;
-        {
-            auto sub = makeSub(env, ep);
-            for (int i = 0; i < N; ++i)
-                send(sub, i);
-            drainAndSettle(env, ep, N);
-        }
-
-        BEAST_EXPECT(ep.received() == N);
-    }
-
-    void
-    testRestartAfterDrain()
-    {
-        testcase("Sending restarts after the queue drains");
-
-        // After a batch drains, sendThread clears mSending and returns.
-        // A later send() must start a fresh sending job; if mSending were
-        // left set (the #6341 failure mode) the second burst would never
-        // be delivered.
-        using namespace jtx;
-        Env env{*this};
-        MockWebhookEndpoint ep;
-
-        {
-            auto sub = makeSub(env, ep);
-
-            // First burst, then wait for the sending job to fully drain
-            // and exit (mSending cleared) — deterministically, not via a
-            // sleep.
-            for (int i = 0; i < 5; ++i)
-                send(sub, i);
-            drainAndSettle(env, ep, 5);
-
-            // Second burst must start a fresh sending job.
-            for (int i = 5; i < 10; ++i)
-                send(sub, i);
-            drainAndSettle(env, ep, 10);
-        }
-
-        BEAST_EXPECT(ep.received() == 10);
-    }
-
-    void
-    testQueueCapDrops()
-    {
-        testcase("Events past the queue cap are dropped");
-
-        // With a tiny cap, pushing far more events than delivery can keep
-        // up with forces send() down the drop path: enqueue is microsecond
-        // -fast while each (delayed) HTTP delivery is a full round-trip, so
-        // the deque sits at the cap and excess events are dropped. The
-        // delay makes "delivery slower than enqueue" hold regardless of
-        // scheduling, so this isn't timing-dependent. We just need some
-        // delivered (cap works) and some dropped (drop path exercised).
-        using namespace jtx;
-        Env env{*this};
-        MockWebhookEndpoint ep;
-        ep.setResponseDelay(50);
-
-        static constexpr int pushed = 50;
-        {
-            auto sub = makeSub(env, ep, /*maxQueueSize*/ 2);
-            for (int i = 0; i < pushed; ++i)
-                send(sub, i);
-            BEAST_EXPECT(waitFor([&] { return sendingIdle(env); }));
-        }
-
-        log << "  queue cap: received " << ep.received() << "/" << pushed
-            << std::endl;
-        BEAST_EXPECT(ep.received() > 0);
-        BEAST_EXPECT(ep.received() < pushed);
-    }
-
-public:
-    void
-    run() override
-    {
-        testDelivery();
-        testErrorsDoNotStall();
-        testRestartAfterDrain();
-        testQueueCapDrops();
-    }
-};
-
-BEAST_DEFINE_TESTSUITE(RPCSub, net, ripple);
-
-}  // namespace test
-}  // namespace ripple

src/test/unit_test/SuiteJournal.h

@@ -19,6 +19,7 @@
 
 #ifndef TEST_UNIT_TEST_SUITE_JOURNAL_H
 #define TEST_UNIT_TEST_SUITE_JOURNAL_H
+#include <xrpl/basics/Log.h>
 #include <xrpl/beast/unit_test.h>
 #include <xrpl/beast/utility/Journal.h>
 #include <mutex>
@@ -31,13 +32,18 @@ class SuiteJournalSink : public beast::Journal::Sink
 {
     std::string partition_;
     beast::unit_test::suite& suite_;
+    Logs* logs_ = nullptr;
 
 public:
     SuiteJournalSink(
         std::string const& partition,
         beast::severities::Severity threshold,
-        beast::unit_test::suite& suite)
-        : Sink(threshold, false), partition_(partition + " "), suite_(suite)
+        beast::unit_test::suite& suite,
+        Logs* logs = nullptr)
+        : Sink(threshold, false)
+        , partition_(partition + " ")
+        , suite_(suite)
+        , logs_(logs)
     {
     }
 
@@ -97,11 +103,12 @@ SuiteJournalSink::writeAlways(
     // Only write the string if the level at least equals the threshold.
     if (level >= threshold())
     {
+        std::string const& output = logs_ ? logs_->applyTransform(text) : text;
         // std::endl flushes → sync() → str()/str("") race in shared buffer →
         // crashes
         static std::mutex log_mutex;
         std::lock_guard lock(log_mutex);
-        suite_.log << s << partition_ << text << std::endl;
+        suite_.log << s << partition_ << output << std::endl;
     }
 }
 

src/xrpld/app/hook/applyHook.h

@@ -12,9 +12,11 @@
 #include <xrpl/protocol/TER.h>
 #include <xrpl/protocol/digest.h>
 #include <any>
+#include <fstream>
 #include <memory>
 #include <optional>
 #include <queue>
+#include <set>
 #include <utility>
 #include <vector>
 #include <wasmedge/wasmedge.h>
@@ -305,6 +307,130 @@ static WasmEdge_String hookFunctionName =
 // see: lib/system/allocator.cpp
 #define WasmEdge_kPageSize 65536ULL
 
+// --- Coverage infrastructure ---
+//
+// Global coverage accumulator keyed by hook hash. Persists across all hook
+// executions in the process. Each __on_source_line call records a (line, col)
+// pair under the executing hook's hash.
+//
+// Test API:
+//   hook::coverageReset()            — clear all accumulated data
+//   hook::coverageHits(hookHash)     — get hits for a specific hook
+//   hook::coverageLabel(hash, label) — register a human-readable label
+//   hook::coverageDump(path)         — write all data to a file
+//
+// The dump file format is:
+//   [label or hash]
+//   hits=<line:col>,<line:col>,...
+
+struct CoverageData
+{
+    std::set<uint32_t> hits{};
+};
+
+// Global accumulator — survives across HookContext lifetimes
+inline std::map<ripple::uint256, CoverageData>&
+coverageMap()
+{
+    static std::map<ripple::uint256, CoverageData> map;
+    return map;
+}
+
+// Hash → label mapping (e.g. hash → "file:tipbot/tip.c")
+inline std::map<ripple::uint256, std::string>&
+coverageLabels()
+{
+    static std::map<ripple::uint256, std::string> labels;
+    return labels;
+}
+
+inline void
+coverageReset()
+{
+    coverageMap().clear();
+    coverageLabels().clear();
+}
+
+inline void
+coverageLabel(ripple::uint256 const& hookHash, std::string const& label)
+{
+    coverageLabels()[hookHash] = label;
+}
+
+inline std::set<uint32_t> const*
+coverageHits(ripple::uint256 const& hookHash)
+{
+    auto& map = coverageMap();
+    auto it = map.find(hookHash);
+    if (it == map.end())
+        return nullptr;
+    return &it->second.hits;
+}
+
+inline bool
+coverageDump(std::string const& path)
+{
+    auto& map = coverageMap();
+    if (map.empty())
+        return false;
+
+    auto& labels = coverageLabels();
+
+    std::ofstream out(path);
+    if (!out)
+        return false;
+
+    for (auto const& [hash, data] : map)
+    {
+        auto it = labels.find(hash);
+        if (it != labels.end())
+            out << "[" << it->second << "]\n";
+        else
+            out << "[" << to_string(hash) << "]\n";
+
+        out << "hits=";
+        bool first = true;
+        for (auto key : data.hits)
+        {
+            if (!first)
+                out << ",";
+            out << (key >> 16) << ":" << (key & 0xFFFF);
+            first = false;
+        }
+        out << "\n\n";
+    }
+
+    return true;
+}
+
+// --- Coverage host callback ---
+
+inline WasmEdge_Result
+onSourceLine(
+    void* data_ptr,
+    const WasmEdge_CallingFrameContext* frameCtx,
+    const WasmEdge_Value* in,
+    WasmEdge_Value* out)
+{
+    // Called by hookz-instrumented WASM at each DWARF source location.
+    // in[0] = line number, in[1] = column number.
+    (void)out;
+    (void)frameCtx;
+    auto* hookCtx = reinterpret_cast<HookContext*>(data_ptr);
+    if (!hookCtx)
+        return WasmEdge_Result_Success;
+
+    uint32_t line = WasmEdge_ValueGetI32(in[0]);
+    uint32_t col = WasmEdge_ValueGetI32(in[1]);
+
+    // Pack (line, col) into a single uint32_t key.
+    // Limits: line < 65536, col < 65536 — more than sufficient for hooks.
+    uint32_t key = (line << 16) | (col & 0xFFFF);
+    coverageMap()[hookCtx->result.hookHash].hits.insert(key);
+
+    return WasmEdge_Result_Success;
+}
+
 /**
  * HookExecutor is effectively a two-part function:
  * The first part sets up the Hook Api inside the wasm import, ready for use
@@ -483,6 +609,22 @@ public:
 #undef HOOK_WRAP_PARAMS
 #pragma pop_macro("HOOK_API_DEFINITION")
 
+        // Coverage callback: void __on_source_line(i32 line, i32 col)
+        // Registered unconditionally — production hooks don't import it,
+        // so it's harmless. Instrumented hooks call it at each DWARF
+        // source location to record line:col coverage hits.
+        {
+            static WasmEdge_ValType paramsOSL[] = {
+                WasmEdge_ValType_I32, WasmEdge_ValType_I32};
+            static auto* ftOSL =
+                WasmEdge_FunctionTypeCreate(paramsOSL, 2, nullptr, 0);
+            auto* hfOSL = WasmEdge_FunctionInstanceCreate(
+                ftOSL, hook::onSourceLine, (void*)(&ctx), 0);
+            static auto nameOSL =
+                WasmEdge_StringCreateByCString("__on_source_line");
+            WasmEdge_ModuleInstanceAddFunction(importObj, nameOSL, hfOSL);
+        }
+
         WasmEdge_TableInstanceContext* hostTable =
             WasmEdge_TableInstanceCreate(tableType);
         WasmEdge_ModuleInstanceAddTable(importObj, tableName, hostTable);

src/xrpld/app/hook/detail/applyHook.cpp

@@ -1111,7 +1111,7 @@ DEFINE_HOOK_FUNCTION(
     if (NOT_IN_BOUNDS(read_ptr, read_len, memory_length))
         return OUT_OF_BOUNDS;
 
-    if (!j.trace())
+    if (!jh.trace())
         return 0ULL;
 
     if (read_len > 128)
@@ -1125,16 +1125,16 @@ DEFINE_HOOK_FUNCTION(
 
         if (read_len > 0)
         {
-            j.trace() << "HookTrace[" << HC_ACC() << "]: "
-                      << std::string_view(
-                             (const char*)memory + read_ptr, read_len)
-                      << ": " << number;
+            JLOG(jh.trace()) << "HookTrace[" << HC_ACC() << "]: "
+                             << std::string_view(
+                                    (const char*)memory + read_ptr, read_len)
+                             << ": " << number;
 
             return 0ULL;
         }
     }
 
-    j.trace() << "HookTrace[" << HC_ACC() << "]: " << number;
+    JLOG(jh.trace()) << "HookTrace[" << HC_ACC() << "]: " << number;
     return 0ULL;
     HOOK_TEARDOWN();
 }
@@ -1154,7 +1154,7 @@ DEFINE_HOOK_FUNCTION(
         NOT_IN_BOUNDS(dread_ptr, dread_len, memory_length))
         return OUT_OF_BOUNDS;
 
-    if (!j.trace())
+    if (!jh.trace())
         return 0ULL;
 
     if (mread_len > 128)
@@ -1214,8 +1214,8 @@ DEFINE_HOOK_FUNCTION(
 
     if (out_len > 0)
     {
-        j.trace() << "HookTrace[" << HC_ACC() << "]: "
-                  << std::string_view((const char*)output_storage, out_len);
+        JLOG(jh.trace()) << "HookTrace[" << HC_ACC() << "]: "
+                         << std::string_view((const char*)output_storage, out_len);
     }
 
     return 0ULL;
@@ -3403,7 +3403,7 @@ DEFINE_HOOK_FUNCTION(
     if (NOT_IN_BOUNDS(read_ptr, read_len, memory_length))
         return OUT_OF_BOUNDS;
 
-    if (!j.trace())
+    if (!jh.trace())
         return 0ULL;
 
     if (read_len > 128)
@@ -3420,8 +3420,8 @@ DEFINE_HOOK_FUNCTION(
 
     if (float1 == 0)
     {
-        j.trace() << "HookTrace[" << HC_ACC() << "]: " << messageKey
-                  << ": Float 0*10^(0) <ZERO>";
+        JLOG(jh.trace()) << "HookTrace[" << HC_ACC() << "]: " << messageKey
+                         << ": Float 0*10^(0) <ZERO>";
         return 0ULL;
     }
 
@@ -3432,14 +3432,14 @@ DEFINE_HOOK_FUNCTION(
         man.value() > maxMantissa || exp.value() < minExponent ||
         exp.value() > maxExponent)
     {
-        j.trace() << "HookTrace[" << HC_ACC() << "]: " << messageKey
-                  << ": Float <INVALID>";
+        JLOG(jh.trace()) << "HookTrace[" << HC_ACC() << "]: " << messageKey
+                         << ": Float <INVALID>";
         return 0ULL;
     }
 
-    j.trace() << "HookTrace[" << HC_ACC() << "]:" << messageKey << ": Float "
-              << (neg ? "-" : "") << man.value() << "*10^(" << exp.value()
-              << ")";
+    JLOG(jh.trace()) << "HookTrace[" << HC_ACC() << "]:" << messageKey
+                     << ": Float " << (neg ? "-" : "") << man.value() << "*10^("
+                     << exp.value() << ")";
     return 0ULL;
 
     HOOK_TEARDOWN();

src/xrpld/app/tx/detail/SetHook.cpp

@@ -546,7 +546,7 @@ SetHook::validateHookSetEntry(SetHookCtx& ctx, STObject const& hookSetObj)
                 }
 
                 auto result = validateGuards(
-                    hook,  // wasm to verify
+                    hook,
                     logger,
                     hsacc,
                     hook_api::getImportWhitelist(ctx.rules),

src/xrpld/net/RPCSub.h

@@ -22,6 +22,7 @@
 
 #include <xrpld/core/JobQueue.h>
 #include <xrpld/net/InfoSub.h>
+#include <boost/asio/io_service.hpp>
 
 namespace ripple {
 
@@ -38,17 +39,16 @@ protected:
     explicit RPCSub(InfoSub::Source& source);
 };
 
+// VFALCO Why is the io_service needed?
 std::shared_ptr<RPCSub>
 make_RPCSub(
     InfoSub::Source& source,
+    boost::asio::io_service& io_service,
     JobQueue& jobQueue,
     std::string const& strUrl,
     std::string const& strUsername,
     std::string const& strPassword,
-    Logs& logs,
-    // Max events buffered before new ones are dropped. Configurable so
-    // tests can exercise the drop path without queueing the full default.
-    std::size_t maxQueueSize = 16384);
+    Logs& logs);
 
 }  // namespace ripple
 

src/xrpld/net/detail/HTTPClient.cpp

@@ -122,20 +122,12 @@ public:
         mComplete = complete;
         mTimeout = timeout;
 
-        // Bind a non-owning `this` (not shared_from_this()) into mBuild.
-        // mBuild is a member, so capturing a shared_ptr to self here would
-        // form a reference cycle (this -> mBuild -> shared_ptr<this>) that
-        // never breaks, leaking the object and its socket FD after the
-        // request completes. mBuild is only ever invoked from
-        // handleRequest(), which always runs inside an async handler that
-        // already holds a shared_from_this(), so the object is guaranteed
-        // alive whenever mBuild fires — a raw `this` is safe.
         request(
             bSSL,
             deqSites,
             std::bind(
                 &HTTPClientImp::makeGet,
-                this,
+                shared_from_this(),
                 strPath,
                 std::placeholders::_1,
                 std::placeholders::_2),
@@ -401,12 +393,8 @@ public:
         if (boost::regex_match(strHeader, smMatch, reBody))  // we got some body
             mBody = smMatch[1];
 
-        bool const hasContentLength =
-            boost::regex_match(strHeader, smMatch, reSize);
-        mReceivedContentLength = hasContentLength;
-
         std::size_t const responseSize = [&] {
-            if (hasContentLength)
+            if (boost::regex_match(strHeader, smMatch, reSize))
                 return beast::lexicalCast<std::size_t>(
                     std::string(smMatch[1]), maxResponseSize_);
             return maxResponseSize_;
@@ -457,24 +445,22 @@ public:
             JLOG(j_.trace()) << "Read error: " << mShutdown.message();
 
             invokeComplete(mShutdown);
-            return;
         }
-
-        // Either the read completed normally or it ended at EOF. EOF is a
-        // successful completion for EOF-delimited responses, but it is an
-        // error when the server promised a Content-Length and closed early.
-        JLOG(j_.trace()) << "Complete.";
-
-        mResponse.commit(bytes_transferred);
-        std::string strBody{
-            {std::istreambuf_iterator<char>(&mResponse)},
-            std::istreambuf_iterator<char>()};
-
-        auto completeEc = ecResult;
-        if (completeEc == boost::asio::error::eof && !mReceivedContentLength)
-            completeEc.clear();
-
-        invokeComplete(completeEc, mStatus, mBody + strBody);
+        else
+        {
+            if (mShutdown)
+            {
+                JLOG(j_.trace()) << "Complete.";
+            }
+            else
+            {
+                mResponse.commit(bytes_transferred);
+                std::string strBody{
+                    {std::istreambuf_iterator<char>(&mResponse)},
+                    std::istreambuf_iterator<char>()};
+                invokeComplete(ecResult, mStatus, mBody + strBody);
+            }
+        }
     }
 
     // Call cancel the deadline timer and invoke the completion routine.
@@ -530,7 +516,6 @@ private:
     boost::asio::streambuf mHeader;
     boost::asio::streambuf mResponse;
     std::string mBody;
-    bool mReceivedContentLength = false;
     const unsigned short mPort;
     std::size_t const maxResponseSize_;
     int mStatus;

src/xrpld/net/detail/RPCCall.cpp

@@ -1585,10 +1585,6 @@ struct RPCCallImp
             // callbackFuncP.
 
             // Receive reply
-            if (ecResult)
-                Throw<std::runtime_error>(
-                    "RPC transport error: " + ecResult.message());
-
             if (strData.empty())
                 Throw<std::runtime_error>(
                     "no response from server. Please "
@@ -1752,7 +1748,6 @@ rpcClient(
             }
 
             {
-                //@@start blocking-request
                 boost::asio::io_service isService;
                 RPCCall::fromNetwork(
                     isService,
@@ -1776,7 +1771,6 @@ rpcClient(
                     headers);
                 isService.run();  // This blocks until there are no more
                                   // outstanding async calls.
-                //@@end blocking-request
             }
             if (jvOutput.isMember("result"))
             {
@@ -1887,21 +1881,15 @@ fromNetwork(
 
     // Send request
 
-    // Number of bytes to try to receive if no Content-Length header is
-    // received. Webhook event deliveries ("event") ignore the response
-    // body, so a missing Content-Length must not pre-allocate the full
-    // 256MB RPC reply budget per in-flight delivery (maxInFlight can be
-    // 32 -> 8GB). Cap those small; genuine RPC replies (CLI) keep the
-    // large budget.
-    auto const RPC_REPLY_MAX_BYTES =
-        (strMethod == "event") ? megabytes(1) : megabytes(256);
+    // Number of bytes to try to receive if no
+    // Content-Length header received
+    constexpr auto RPC_REPLY_MAX_BYTES = megabytes(256);
 
     using namespace std::chrono_literals;
     // auto constexpr RPC_NOTIFY = 10min; // Wietse: lolwut 10 minutes for one
     // HTTP call?
     auto constexpr RPC_NOTIFY = 30s;
 
-    //@@start async-request
     HTTPClient::request(
         bSSL,
         io_service,
@@ -1926,7 +1914,6 @@ fromNetwork(
             std::placeholders::_3,
             j),
         j);
-    //@@end async-request
 }
 
 }  // namespace RPCCall

src/xrpld/net/detail/RPCSub.cpp

@@ -24,30 +24,29 @@
 #include <xrpl/basics/contract.h>
 #include <xrpl/json/to_string.h>
 #include <deque>
-#include <memory>
 
 namespace ripple {
 
 // Subscription object for JSON-RPC
-class RPCSubImp : public RPCSub, public std::enable_shared_from_this<RPCSubImp>
+class RPCSubImp : public RPCSub
 {
 public:
     RPCSubImp(
         InfoSub::Source& source,
+        boost::asio::io_service& io_service,
         JobQueue& jobQueue,
         std::string const& strUrl,
         std::string const& strUsername,
         std::string const& strPassword,
-        Logs& logs,
-        std::size_t maxQueueSize)
+        Logs& logs)
         : RPCSub(source)
+        , m_io_service(io_service)
         , m_jobQueue(jobQueue)
         , mUrl(strUrl)
         , mSSL(false)
         , mUsername(strUsername)
         , mPassword(strPassword)
         , mSending(false)
-        , maxQueueSize_(maxQueueSize)
         , j_(logs.journal("RPCSub"))
         , logs_(logs)
     {
@@ -79,26 +78,14 @@ public:
     {
         std::lock_guard sl(mLock);
 
-        if (mDeque.size() >= maxQueueSize_)
-        {
-            // Always advance mSeq so consumers can detect the gap, but
-            // rate-limit the log: a hopelessly behind endpoint drops on
-            // every send() and would otherwise flood the log. Warn on
-            // the first drop of a run and then once per dropLogInterval.
-            if (mDropped++ % dropLogInterval == 0)
-            {
-                JLOG(j_.warn())
-                    << "RPCCall::fromNetwork drop: queue full ("
-                    << mDeque.size() << "), seq=" << mSeq
-                    << ", endpoint=" << mIp << ", dropped=" << mDropped;
-            }
-            ++mSeq;
-            return;
-        }
-
-        // Endpoint caught up enough to accept again; reset so the next
-        // overflow burst logs its first drop immediately.
-        mDropped = 0;
+        // Wietse: we're not going to limit this, this is admin-port only, scale
+        // accordingly Dropping events just like this results in inconsistent
+        // data on the receiving end if (mDeque.size() >= eventQueueMax)
+        // {
+        //     // Drop the previous event.
+        //     JLOG(j_.warn()) << "RPCCall::fromNetwork drop";
+        //     mDeque.pop_back();
+        // }
 
         auto jm = broadcast ? j_.debug() : j_.info();
         JLOG(jm) << "RPCCall::fromNetwork push: " << jvObj;
@@ -110,7 +97,10 @@ public:
             // Start a sending thread.
             JLOG(j_.info()) << "RPCCall::fromNetwork start";
 
-            startSendingJob();
+            mSending = m_jobQueue.addJob(
+                jtCLIENT_SUBSCRIBE, "RPCSub::sendThread", [this]() {
+                    sendThread();
+                });
         }
     }
 
@@ -131,66 +121,48 @@ public:
     }
 
 private:
-    // Maximum concurrent HTTP deliveries per batch. Bounds file
-    // descriptor usage while still allowing parallel delivery to
-    // capable endpoints. With a 1024 FD process limit shared across
-    // peers, clients, and the node store, 32 per subscriber is a
-    // meaningful but survivable chunk even with multiple subscribers.
-    static constexpr int maxInFlight = 32;
-
-    // Log one drop warning per this many drops while the queue stays
-    // full, to avoid flooding the log on a persistently behind endpoint.
-    static constexpr std::size_t dropLogInterval = 1000;
-
-    // Schedule a sending job. Must be called under mLock. The job holds a
-    // weak_ptr and re-locks it on entry, so the RPCSub is kept alive for
-    // the duration of the batch even if it is unsubscribed (and would
-    // otherwise be destroyed) concurrently — sendThread dereferences this
-    // only via that strong ref. mDeque events are delivered until the sub
-    // is gone, after which weak.lock() fails and the job is a no-op.
-    void
-    startSendingJob()
-    {
-        std::weak_ptr<RPCSubImp> weak = weak_from_this();
-        mSending = m_jobQueue.addJob(
-            jtCLIENT_SUBSCRIBE, "RPCSub::sendThread", [weak]() {
-                if (auto self = weak.lock())
-                    self->sendThread();
-            });
-    }
-
+    // XXX Could probably create a bunch of send jobs in a single get of the
+    // lock.
     void
     sendThread()
     {
-        // Process exactly ONE batch per job, then re-queue if more events
-        // remain, rather than draining the whole backlog in a single job.
-        // A local io_service's .run() blocks this worker thread for the
-        // batch (up to the per-request timeout), so re-queueing between
-        // batches keeps one slow/hung subscriber from monopolising a
-        // job-queue worker and starving consensus/ledger/RPC work.
-        //
-        // mSending must be cleared under the lock on every non-requeue
-        // exit path; if it ever stays set without a job in flight, send()
-        // sees mSending == true and never restarts us, stalling the queue
-        // forever — the original bug (xrpld issue #6341).
-        boost::asio::io_service io_service;
-        int dispatched = 0;
+        Json::Value jvEvent;
+        bool bSend;
 
-        try
+        do
         {
             {
+                // Obtain the lock to manipulate the queue and change sending.
                 std::lock_guard sl(mLock);
 
-                while (!mDeque.empty() && dispatched < maxInFlight)
+                if (mDeque.empty())
+                {
+                    mSending = false;
+                    bSend = false;
+                }
+                else
                 {
                     auto const [seq, env] = mDeque.front();
+
                     mDeque.pop_front();
 
-                    Json::Value jvEvent = env;
+                    jvEvent = env;
                     jvEvent["seq"] = seq;
 
+                    bSend = true;
+                }
+            }
+
+            // Send outside of the lock.
+            if (bSend)
+            {
+                // XXX Might not need this in a try.
+                try
+                {
+                    JLOG(j_.info()) << "RPCCall::fromNetwork: " << mIp;
+
                     RPCCall::fromNetwork(
-                        io_service,
+                        m_io_service,
                         mIp,
                         mPort,
                         mUsername,
@@ -201,51 +173,21 @@ private:
                         mSSL,
                         true,
                         logs_);
-                    ++dispatched;
+                }
+                catch (const std::exception& e)
+                {
+                    JLOG(j_.info())
+                        << "RPCCall::fromNetwork exception: " << e.what();
                 }
             }
-
-            // dispatched is always > 0 here (send() only starts a job
-            // after enqueuing, and the re-queue below only fires with a
-            // non-empty deque), but guard anyway so an empty batch can't
-            // log/spin — it falls straight through to clear mSending.
-            if (dispatched > 0)
-            {
-                JLOG(j_.info()) << "RPCCall::fromNetwork: " << mIp
-                                << " dispatching " << dispatched << " events";
-
-                io_service.run();
-            }
-        }
-        catch (std::exception const& e)
-        {
-            // Bail rather than re-queue: a persistently failing endpoint
-            // would otherwise spin the job queue. mSending is reset so the
-            // next send() restarts delivery.
-            JLOG(j_.warn()) << "RPCSub::sendThread exception: " << e.what();
-            std::lock_guard sl(mLock);
-            mSending = false;
-            return;
-        }
-        catch (...)
-        {
-            JLOG(j_.warn()) << "RPCSub::sendThread unknown exception";
-            std::lock_guard sl(mLock);
-            mSending = false;
-            return;
-        }
-
-        // Batch complete: re-queue for the next one (mSending stays set)
-        // or clear mSending if the queue drained — both under the lock to
-        // avoid a lost-wakeup race with send().
-        std::lock_guard sl(mLock);
-        if (mDeque.empty())
-            mSending = false;
-        else
-            startSendingJob();
+        } while (bSend);
     }
 
 private:
+    // Wietse: we're not going to limit this, this is admin-port only, scale
+    // accordingly enum { eventQueueMax = 32 };
+
+    boost::asio::io_service& m_io_service;
     JobQueue& m_jobQueue;
 
     std::string mUrl;
@@ -258,15 +200,8 @@ private:
 
     int mSeq;  // Next id to allocate.
 
-    std::size_t mDropped = 0;  // Consecutive drops while queue is full.
-
     bool mSending;  // Sending threead is active.
 
-    // Maximum queued events before dropping. The default (16384) is a
-    // ~10-minute buffer at 100+ events/ledger; a hopelessly behind
-    // endpoint trips it and consumers detect the gap via the seq field.
-    std::size_t const maxQueueSize_;
-
     std::deque<std::pair<int, Json::Value>> mDeque;
 
     beast::Journal const j_;
@@ -282,21 +217,21 @@ RPCSub::RPCSub(InfoSub::Source& source) : InfoSub(source, Consumer())
 std::shared_ptr<RPCSub>
 make_RPCSub(
     InfoSub::Source& source,
+    boost::asio::io_service& io_service,
     JobQueue& jobQueue,
     std::string const& strUrl,
     std::string const& strUsername,
     std::string const& strPassword,
-    Logs& logs,
-    std::size_t maxQueueSize)
+    Logs& logs)
 {
     return std::make_shared<RPCSubImp>(
         std::ref(source),
+        std::ref(io_service),
         std::ref(jobQueue),
         strUrl,
         strUsername,
         strPassword,
-        logs,
-        maxQueueSize);
+        logs);
 }
 
 }  // namespace ripple

src/xrpld/rpc/handlers/Subscribe.cpp

@@ -76,6 +76,7 @@ doSubscribe(RPC::JsonContext& context)
             {
                 auto rspSub = make_RPCSub(
                     context.app.getOPs(),
+                    context.app.getIOService(),
                     context.app.getJobQueue(),
                     strUrl,
                     strUsername,