From 2d7b17ae1f023d2818cf870bd1786afbdfec0c0f Mon Sep 17 00:00:00 2001 From: Nicholas Dudfield Date: Wed, 10 Jun 2026 16:45:52 +0700 Subject: [PATCH] test: harden RPCSub suite for coverage build; cover the drop path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The RPCSub tests crashed/timed out under the instrumented Debug coverage build (where the full HTTP stack is ~50x slower than Release): - Deterministic teardown: wait on JobQueue::getJobCountTotal( jtCLIENT_SUBSCRIBE) == 0 before destroying the sub. sendThread captures a raw `this`; a fixed-sleep grace raced it under slow builds → UAF. - Lower per-test event counts (50 -> 10) so a handful of real HTTP deliveries finish well within the timeout under instrumentation. Queue-cap drop coverage: make_RPCSub gains a defaulted maxQueueSize parameter (production default unchanged at 16384) so a test can set a tiny cap and exercise the drop path by out-pacing delivery — enqueue is microsecond-fast while each delivery is a full round-trip — instead of queueing 16384 events. New testQueueCapDrops confirms some events are delivered and some dropped. Raises RPCSub.cpp patch coverage 62% -> 87% (overall diff 65% -> 87%); the only remaining uncovered lines are the defensive sendThread exception catches. --- src/test/net/RPCSub_test.cpp | 99 ++++++++++++++++++++++++--------- src/xrpld/net/RPCSub.h | 5 +- src/xrpld/net/detail/RPCSub.cpp | 24 ++++---- 3 files changed, 91 insertions(+), 37 deletions(-) diff --git a/src/test/net/RPCSub_test.cpp b/src/test/net/RPCSub_test.cpp index 0a4e4c826..92de618e4 100644 --- a/src/test/net/RPCSub_test.cpp +++ b/src/test/net/RPCSub_test.cpp @@ -18,6 +18,8 @@ //============================================================================== #include +#include +#include #include #include @@ -139,11 +141,11 @@ private: 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 bool - waitFor( - Cond cond, - std::chrono::milliseconds timeout = std::chrono::seconds{10}) + 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) @@ -152,7 +154,10 @@ class RPCSub_test : public beast::unit_test::suite } std::shared_ptr - makeSub(jtx::Env& env, MockWebhookEndpoint& ep) + makeSub( + jtx::Env& env, + MockWebhookEndpoint& ep, + std::size_t maxQueueSize = 16384) { return make_RPCSub( env.app().getOPs(), @@ -160,17 +165,32 @@ class RPCSub_test : public beast::unit_test::suite "http://127.0.0.1:" + std::to_string(ep.port()) + "/", "", "", - env.app().logs()); + env.app().logs(), + maxQueueSize); } - // Wait until all queued events have been delivered, then give the - // sending job a moment to finish. sendThread captures a raw `this`, - // so the RPCSub must not be destroyed while it is still running. - void - drainAndSettle(MockWebhookEndpoint& ep, int expected) + // 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) { - BEAST_EXPECT(waitFor([&] { return ep.received() >= expected; })); - std::this_thread::sleep_for(std::chrono::milliseconds(300)); + 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 @@ -186,18 +206,16 @@ class RPCSub_test : public beast::unit_test::suite { testcase("Webhook events are delivered"); - // N > maxInFlight(32) so sendThread drains in multiple batches - // within a single invocation (exercises the dispatch loop). using namespace jtx; Env env{*this}; MockWebhookEndpoint ep; - static constexpr int N = 50; + static constexpr int N = 10; { auto sub = makeSub(env, ep); for (int i = 0; i < N; ++i) send(sub, i); - drainAndSettle(ep, N); + drainAndSettle(env, ep, N); } BEAST_EXPECT(ep.received() == N); @@ -217,12 +235,12 @@ class RPCSub_test : public beast::unit_test::suite MockWebhookEndpoint ep; ep.setStatus(500); - static constexpr int N = 50; + static constexpr int N = 10; { auto sub = makeSub(env, ep); for (int i = 0; i < N; ++i) send(sub, i); - drainAndSettle(ep, N); + drainAndSettle(env, ep, N); } BEAST_EXPECT(ep.received() == N); @@ -244,18 +262,48 @@ class RPCSub_test : public beast::unit_test::suite { auto sub = makeSub(env, ep); - for (int i = 0; i < 20; ++i) + // 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); - BEAST_EXPECT(waitFor([&] { return ep.received() >= 20; })); - // Let the first sending job finish so mSending goes false. - std::this_thread::sleep_for(std::chrono::milliseconds(200)); + drainAndSettle(env, ep, 5); - for (int i = 20; i < 40; ++i) + // Second burst must start a fresh sending job. + for (int i = 5; i < 10; ++i) send(sub, i); - drainAndSettle(ep, 40); + drainAndSettle(env, ep, 10); } - BEAST_EXPECT(ep.received() == 40); + 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 HTTP delivery is a full round-trip, so the deque + // sits at the cap and excess events are dropped. We just need some + // delivered (cap works) and some dropped (drop path exercised). + using namespace jtx; + Env env{*this}; + MockWebhookEndpoint ep; + + 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: @@ -265,6 +313,7 @@ public: testDelivery(); testErrorsDoNotStall(); testRestartAfterDrain(); + testQueueCapDrops(); } }; diff --git a/src/xrpld/net/RPCSub.h b/src/xrpld/net/RPCSub.h index 3cf7f72c8..5cf49efbf 100644 --- a/src/xrpld/net/RPCSub.h +++ b/src/xrpld/net/RPCSub.h @@ -45,7 +45,10 @@ make_RPCSub( std::string const& strUrl, std::string const& strUsername, std::string const& strPassword, - Logs& logs); + 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); } // namespace ripple diff --git a/src/xrpld/net/detail/RPCSub.cpp b/src/xrpld/net/detail/RPCSub.cpp index 87d9fbca5..e6f76aed0 100644 --- a/src/xrpld/net/detail/RPCSub.cpp +++ b/src/xrpld/net/detail/RPCSub.cpp @@ -37,7 +37,8 @@ public: std::string const& strUrl, std::string const& strUsername, std::string const& strPassword, - Logs& logs) + Logs& logs, + std::size_t maxQueueSize) : RPCSub(source) , m_jobQueue(jobQueue) , mUrl(strUrl) @@ -45,6 +46,7 @@ public: , mUsername(strUsername) , mPassword(strPassword) , mSending(false) + , maxQueueSize_(maxQueueSize) , j_(logs.journal("RPCSub")) , logs_(logs) { @@ -76,7 +78,7 @@ public: { std::lock_guard sl(mLock); - if (mDeque.size() >= maxQueueSize) + if (mDeque.size() >= maxQueueSize_) { // Always advance mSeq so consumers can detect the gap, but // rate-limit the log: a hopelessly behind endpoint drops on @@ -138,13 +140,6 @@ private: // meaningful but survivable chunk even with multiple subscribers. static constexpr int maxInFlight = 32; - // Maximum queued events before dropping. At ~5-10KB per event - // this is ~80-160MB worst case — trivial memory-wise. The real - // purpose is detecting a hopelessly behind endpoint: at 100+ - // events per ledger (every ~4s), 16384 events is ~10 minutes - // of buffer. Consumers detect gaps via the seq field. - static constexpr std::size_t maxQueueSize = 16384; - // 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; @@ -255,6 +250,11 @@ private: 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> mDeque; beast::Journal const j_; @@ -274,7 +274,8 @@ make_RPCSub( std::string const& strUrl, std::string const& strUsername, std::string const& strPassword, - Logs& logs) + Logs& logs, + std::size_t maxQueueSize) { return std::make_shared( std::ref(source), @@ -282,7 +283,8 @@ make_RPCSub( strUrl, strUsername, strPassword, - logs); + logs, + maxQueueSize); } } // namespace ripple