ARCHIVE/xahaud
3
0
Fork 0
mirror of https://github.com/Xahau/xahaud.git synced 2026-07-30 18:40:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

test: harden RPCSub suite for coverage build; cover the drop path

Browse Source 
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.
This commit is contained in:
Nicholas Dudfield
2026-06-10 16:45:52 +07:00
parent 159e00570d
commit 2d7b17ae1f
3 changed files with 91 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

99
src/test/net/RPCSub_test.cpp
View File

@@ -18,6 +18,8 @@
//==============================================================================
#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>
@@ -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 <class Cond>
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<RPCSub>
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();
}
};

5
src/xrpld/net/RPCSub.h
View File

@@ -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

24
src/xrpld/net/detail/RPCSub.cpp
View File

@@ -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<std::pair<int, Json::Value>> 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<RPCSubImp>(
std::ref(source),
@@ -282,7 +283,8 @@ make_RPCSub(
strUrl,
strUsername,
strPassword,
logs);
logs,
maxQueueSize);
}
} // namespace ripple