ARCHIVE/rippled
1
0
Fork 0
mirror of https://github.com/XRPLF/rippled.git synced 2026-07-27 17:10:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/develop' into pratik/std-coro/add-coroutine-primitives

Browse Source
This commit is contained in:
Pratik Mankawde
2026-07-27 13:47:54 +01:00
parent 9bddc54722 20801d98ac
commit 5ea72b45f1
18 changed files with 321 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.github/workflows/check-pr-description.yml vendored
View File

@@ -23,7 +23,7 @@ jobs:
runs-on: ubuntu-latest
steps:
- name: Checkout repository
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
- name: Write PR body to file
env:

4
.github/workflows/check-tools.yml vendored
View File

@@ -47,7 +47,7 @@ jobs:
tag: ${{ steps.tag.outputs.tag }}
steps:
- name: Checkout repository
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
- name: Read nix image tag
id: tag
@@ -76,7 +76,7 @@ jobs:
container: ${{ !matrix.nix_develop && format('ghcr.io/xrplf/xrpld/nix-ubuntu:{0}', needs.linux-image-tag.outputs.tag) || null }}
steps:
- name: Checkout repository
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
- name: Prepare runner
uses: XRPLF/actions/prepare-runner@e4b6449d55a61c002d7c3fdfa6c20f721ede0606

2
.github/workflows/on-pr.yml vendored
View File

@@ -52,7 +52,7 @@ jobs:
runs-on: ubuntu-latest
steps:
- name: Checkout repository
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
- name: Determine changed files
# This step checks whether any files have changed that should
# cause the next jobs to run. We do it this way rather than

2
.github/workflows/publish-docs.yml vendored
View File

@@ -44,7 +44,7 @@ jobs:
container: ghcr.io/xrplf/xrpld/nix-ubuntu:sha-40cdf49
steps:
- name: Checkout repository
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
- name: Prepare runner
uses: XRPLF/actions/prepare-runner@e4b6449d55a61c002d7c3fdfa6c20f721ede0606

2
.github/workflows/reusable-build-test-config.yml vendored
View File

@@ -110,7 +110,7 @@ jobs:
uses: XRPLF/actions/cleanup-workspace@c7d9ce5ebb03c752a354889ecd870cadfc2b1cd4
- name: Checkout repository
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
- name: Prepare runner
uses: XRPLF/actions/prepare-runner@e4b6449d55a61c002d7c3fdfa6c20f721ede0606

2
.github/workflows/reusable-check-levelization.yml vendored
View File

@@ -18,7 +18,7 @@ jobs:
runs-on: ubuntu-latest
steps:
- name: Checkout repository
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
- name: Check levelization
run: python .github/scripts/levelization/generate.py
- name: Check for differences

2
.github/workflows/reusable-check-rename.yml vendored
View File

@@ -18,7 +18,7 @@ jobs:
runs-on: ubuntu-latest
steps:
- name: Checkout repository
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
- name: Check definitions
run: .github/scripts/rename/definitions.sh .
- name: Check copyright notices

2
.github/workflows/reusable-clang-tidy.yml vendored
View File

@@ -40,7 +40,7 @@ jobs:
issues: write
steps:
- name: Checkout repository
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
- name: Prepare runner
uses: XRPLF/actions/prepare-runner@e4b6449d55a61c002d7c3fdfa6c20f721ede0606

4
.github/workflows/reusable-package.yml vendored
View File

@@ -27,7 +27,7 @@ jobs:
matrix: ${{ steps.generate.outputs.matrix }}
steps:
- name: Checkout repository
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
- name: Set up Python
uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
@@ -54,7 +54,7 @@ jobs:
steps:
- name: Checkout repository
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
- name: Download pre-built binary
uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1

2
.github/workflows/reusable-strategy-matrix.yml vendored
View File

@@ -23,7 +23,7 @@ jobs:
matrix: ${{ steps.generate.outputs.matrix }}
steps:
- name: Checkout repository
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
- name: Set up Python
uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0

2
.github/workflows/reusable-upload-recipe.yml vendored
View File

@@ -47,7 +47,7 @@ jobs:
CONAN_PASSWORD_XRPLF: ${{ secrets.remote_password }}
steps:
- name: Checkout repository
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
- name: Generate build version number
id: version

2
.github/workflows/upload-conan-deps.yml vendored
View File

@@ -65,7 +65,7 @@ jobs:
uses: XRPLF/actions/cleanup-workspace@c7d9ce5ebb03c752a354889ecd870cadfc2b1cd4
- name: Checkout repository
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
- name: Prepare runner
uses: XRPLF/actions/prepare-runner@e4b6449d55a61c002d7c3fdfa6c20f721ede0606

11
src/test/jtx/AbstractClient.h
View File

@@ -40,6 +40,17 @@ public:
*/
[[nodiscard]] virtual unsigned
version() const = 0;
/**
* Close the client's connection to the server.
*
* Releases the connection the client holds against the server's per-port
* connection limit. After this call the client must not be used to
* invoke() again. Tests use this to deterministically free the slot
* rather than waiting for the server's idle timeout to drop it.
*/
virtual void
disconnect() = 0;
};
} // namespace xrpl::test

70
src/test/jtx/Env.h
View File

@@ -482,11 +482,52 @@ public:
app().getNumberOfThreads() == 1,
"syncClose() is only useful on an application with a single thread");
auto const result = close();
auto serverBarrier = std::make_shared<std::promise<void>>();
auto future = serverBarrier->get_future();
boost::asio::post(app().getIOContext(), [serverBarrier]() { serverBarrier->set_value(); });
auto const status = future.wait_for(timeout);
return result && status == std::future_status::ready;
return result && drainServerIo(timeout);
}
/**
* Disconnect the Env's built-in client and wait for the server to
* register the dropped connection.
*
* Env holds one persistent client connection to the server's RPC port for
* its whole lifetime (see client()), and that connection counts against
* the port's connection limit. Tests that need a known starting occupancy
* can call this to deterministically release that slot instead of waiting
* out the server's localhost idle timeout.
*
* The server decrements its per-port connection count in the peer's
* destructor, which runs when the io_context processes the end-of-stream
* on the closed socket. After closing the client this drains the server's
* io_context twice: the first barrier guarantees the reactor has reaped
* the closed socket and queued the peer's teardown, and the second
* guarantees that teardown (and therefore the count decrement) has run.
*
* This is only sound when the server uses a single io_context thread, so
* that draining establishes ordering against the teardown - configure the
* Env with singleThreadIo() (as syncClose() also requires). Like
* syncClose(), it relies on loopback teardown latency being negligible.
*
* @param timeout Maximum time to wait for each barrier task to execute
* @return true if both barriers executed within timeout, false otherwise
*/
[[nodiscard]] bool
disconnectClient(std::chrono::steady_clock::duration timeout = std::chrono::seconds{1})
{
XRPL_ASSERT(
app().getNumberOfThreads() == 1,
"disconnectClient() is only useful on an application with a single "
"thread");
bundle_.client->disconnect();
// Drain the server's single io thread twice: the first barrier flushes
// the reactor's reap of the closed socket (queuing the peer teardown),
// the second flushes that teardown - and therefore the connection-count
// decrement. Both run unconditionally so a timed-out first drain does
// not short-circuit the second.
bool const reaped = drainServerIo(timeout);
bool const toreDown = drainServerIo(timeout);
return reaped && toreDown;
}
/**
@@ -846,6 +887,25 @@ public:
}
private:
/**
* Drain the (single) server io_context thread once.
*
* Posts a barrier task to the server's io_context and blocks until it
* runs, so every task queued before it has been processed. Only meaningful
* with a single io thread (see syncClose()/disconnectClient()).
*
* @param timeout Maximum time to wait for the barrier task to execute
* @return true if the barrier ran within timeout, false otherwise
*/
[[nodiscard]] bool
drainServerIo(std::chrono::steady_clock::duration timeout)
{
auto barrier = std::make_shared<std::promise<void>>();
auto future = barrier->get_future();
boost::asio::post(app().getIOContext(), [barrier]() { barrier->set_value(); });
return future.wait_for(timeout) == std::future_status::ready;
}
void
fund(bool setDefaultRipple, STAmount const& amount, Account const& account);

44
src/test/jtx/WSClient_test.cpp
View File

@@ -13,8 +13,9 @@ class WSClient_test : public beast::unit_test::Suite
{
public:
void
run() override
testSmoke()
{
testcase("smoke");
using namespace jtx;
Env env(*this);
auto wsc = makeWSClient(env.app().config());
@@ -28,6 +29,47 @@ public:
auto jv = wsc->getMsg(std::chrono::seconds(1));
pass();
}
void
testGracefulDisconnect()
{
testcase("graceful disconnect");
using namespace jtx;
using namespace std::chrono;
Env env(*this);
auto wsc = makeWSClient(env.app().config());
// Put real traffic on the connection before closing it.
json::Value stream;
stream["streams"] = json::ValueType::Array;
stream["streams"].append("ledger");
auto const sub = wsc->invoke("subscribe", stream);
BEAST_EXPECT(sub.isMember("result") || sub.isMember("status"));
// disconnect() performs a graceful WebSocket closing handshake and
// blocks until the server acknowledges. On loopback that completes in
// well under its internal 1s timeout; only a broken async_close/ack
// coordination would fall through to the force-close path at ~1s. A
// generous bound keeps this from flaking under load while still
// catching that regression.
auto const start = steady_clock::now();
wsc->disconnect();
auto const elapsed = duration_cast<milliseconds>(steady_clock::now() - start);
BEAST_EXPECT(elapsed < milliseconds{750});
// disconnect() must be idempotent: a second call (and the subsequent
// destructor) must not hang, double-close, or crash.
wsc->disconnect();
pass();
}
void
run() override
{
testSmoke();
testGracefulDisconnect();
}
};
BEAST_DEFINE_TESTSUITE(WSClient, jtx, xrpl);

92
src/test/jtx/impl/JSONRPCClient.cpp
View File

@@ -14,18 +14,23 @@
#include <xrpl/server/Port.h>
#include <boost/asio/buffer.hpp>
#include <boost/asio/error.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/ip/address_v4.hpp>
#include <boost/asio/ip/address_v6.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/beast/core/multi_buffer.hpp>
#include <boost/beast/http/dynamic_body.hpp>
#include <boost/beast/http/error.hpp>
#include <boost/beast/http/message.hpp>
#include <boost/beast/http/read.hpp>
#include <boost/beast/http/string_body.hpp>
#include <boost/beast/http/verb.hpp>
#include <boost/beast/http/write.hpp>
#include <boost/system/system_error.hpp>
#include <algorithm>
#include <array>
#include <iostream>
#include <memory>
#include <sstream>
@@ -84,6 +89,40 @@ class JSONRPCClient : public AbstractClient
boost::beast::multi_buffer bout_;
unsigned rpcVersion_;
bool disconnected_ = false;
// Errors that mean the persistent keep-alive connection was dropped by the
// server (rather than a genuine protocol failure), so the request can be
// safely retried on a fresh connection.
static bool
droppedConnection(boost::system::error_code const& ec)
{
namespace error = boost::asio::error;
static auto const kDroppedConnectionErrors = std::to_array<boost::system::error_code>({
boost::beast::http::error::end_of_stream,
error::eof,
error::connection_reset,
error::connection_aborted,
error::broken_pipe,
error::not_connected,
});
return std::ranges::any_of(
kDroppedConnectionErrors,
[&ec](boost::system::error_code const& e) { return ec == e; });
}
// Tear down and re-establish the socket to ep_, discarding any buffered
// bytes left over from the dropped connection.
void
reconnect()
{
boost::system::error_code ec;
stream_.close(ec);
bin_.clear();
stream_.connect(ep_);
}
public:
explicit JSONRPCClient(Config const& cfg, unsigned rpcVersion)
: ep_(getEndpoint(cfg)), stream_(ios_), rpcVersion_(rpcVersion)
@@ -91,12 +130,10 @@ public:
stream_.connect(ep_);
}
/*
Return value is an Object type with up to three keys:
status
error
result
*/
// Return value is an Object type with up to three keys:
// status
// error
// result
json::Value
invoke(std::string const& cmd, json::Value const& params) override
{
@@ -104,6 +141,13 @@ public:
using namespace boost::asio;
using namespace std::string_literals;
// Once disconnect() has released the slot, the client must not be
// reused (see AbstractClient::disconnect). Refuse rather than let the
// failed write/read below trip the reconnect path and silently
// re-consume a connection slot, which would defeat disconnectClient().
if (disconnected_)
Throw<std::logic_error>("JSONRPCClient::invoke called after disconnect()");
request<string_body> req;
req.method(boost::beast::http::verb::post);
req.target("/");
@@ -131,10 +175,29 @@ public:
req.body() = to_string(jr);
}
req.prepare_payload();
write(stream_, req);
// The client keeps a single keep-alive connection for its whole
// lifetime, but the server drops idle localhost connections after a few
// seconds (BaseHTTPPeer::kTimeoutSecondsLocal). If a slow gap between
// requests let the server close the socket, the write/read here fails
// with end_of_stream; reconnect and retry the request exactly once.
response<dynamic_body> res;
read(stream_, bin_, res);
auto writeAndRead = [&] {
write(stream_, req);
read(stream_, bin_, res);
};
try
{
writeAndRead();
}
catch (boost::system::system_error const& e)
{
if (!droppedConnection(e.code()))
throw;
reconnect();
res = {};
writeAndRead();
}
json::Reader jr;
json::Value jv;
@@ -151,6 +214,19 @@ public:
{
return rpcVersion_;
}
void
disconnect() override
{
if (disconnected_)
return;
disconnected_ = true;
boost::system::error_code ec;
stream_.shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec);
stream_.close(ec);
}
};
std::unique_ptr<AbstractClient>

101
src/test/jtx/impl/WSClient.cpp
View File

@@ -2,6 +2,7 @@
#include <xrpld/core/Config.h>
#include <xrpl/basics/Mutex.hpp>
#include <xrpl/basics/contract.h>
#include <xrpl/config/BasicConfig.h>
#include <xrpl/config/Constants.h>
@@ -112,10 +113,11 @@ class WSClientImpl : public WSClient
bool peerClosed_ = false;
// synchronize destructor
bool b0_ = false;
std::mutex m0_;
std::condition_variable cv0_;
// disconnect() waits on this until the read loop ends (for any reason:
// the server acknowledged our close, or a timeout force-closed the socket).
static constexpr auto kDisconnectTimeout = std::chrono::seconds{1};
xrpl::Mutex<bool> readEnded_;
std::condition_variable readEndCv_;
// synchronize message queue
std::mutex m_;
@@ -127,23 +129,26 @@ class WSClientImpl : public WSClient
void
cleanup()
{
boost::asio::post(ios_, boost::asio::bind_executor(strand_, [this] {
if (!peerClosed_)
{
ws_.async_close(
{}, boost::asio::bind_executor(strand_, [&](error_code) {
try
{
stream_.cancel();
}
// NOLINTNEXTLINE(bugprone-empty-catch)
catch (boost::system::system_error const&)
{
// ignored
}
}));
}
}));
boost::asio::post(
ios_, //
boost::asio::bind_executor(strand_, [this] {
if (!peerClosed_)
{
ws_.async_close(
{}, //
boost::asio::bind_executor(strand_, [&](error_code) {
try
{
stream_.cancel();
}
// NOLINTNEXTLINE(bugprone-empty-catch)
catch (boost::system::system_error const&)
{
// ignored
}
}));
}
}));
work_ = std::nullopt;
thread_.join();
}
@@ -289,6 +294,44 @@ public:
return rpcVersion_;
}
void
disconnect() override
{
// Perform a graceful WebSocket closing handshake and block until the
// read loop ends, so the server observes a clean close (not a RST) and
// has finished tearing the connection down by the time we return.
// If the server already closed, the wait below returns immediately.
boost::asio::post(
ios_,
boost::asio::bind_executor(
strand_, //
[this] {
if (!peerClosed_)
{
ws_.async_close(
boost::beast::websocket::close_code::normal,
boost::asio::bind_executor(strand_, [](error_code) {}));
}
}));
auto lock = readEnded_.lock<std::unique_lock>();
readEndCv_.wait_for(lock, kDisconnectTimeout, [&lock] { return *lock; });
// On timeout (server gone or not replying) force the socket closed so
// the outstanding read ends and the worker thread can later be joined.
if (!*lock)
{
boost::asio::post(
ios_,
boost::asio::bind_executor(
strand_, //
[this] {
boost::system::error_code ec;
stream_.close(ec);
}));
}
}
private:
void
onReadMsg(error_code const& ec)
@@ -297,33 +340,31 @@ private:
{
if (ec == boost::beast::websocket::error::closed)
peerClosed_ = true;
*readEnded_.lock() = true;
readEndCv_.notify_all();
return;
}
json::Value jv;
json::Reader jr;
jr.parse(bufferString(rb_.data()), jv);
rb_.consume(rb_.size());
auto m = std::make_shared<Msg>(std::move(jv));
{
std::scoped_lock const lock(m_);
msgs_.push_front(m);
cv_.notify_all();
}
ws_.async_read(
rb_, boost::asio::bind_executor(strand_, [this](error_code const& ec, std::size_t) {
onReadMsg(ec);
}));
}
// Called when the read op terminates
void
onReadDone()
{
std::scoped_lock const lock(m0_);
b0_ = true;
cv0_.notify_all();
}
};
std::unique_ptr<WSClient>

48
src/test/server/ServerStatus_test.cpp
View File

@@ -558,10 +558,12 @@ class ServerStatus_test : public beast::unit_test::Suite, public beast::test::En
using namespace test::jtx;
using namespace boost::asio;
using namespace boost::beast::http;
Env env{*this, envconfig([&](std::unique_ptr<Config> cfg) {
// Run the server with a single io thread so disconnectClient() below
// can deterministically drain the server's io_context (see its docs).
Env env{*this, singleThreadIo(envconfig([&](std::unique_ptr<Config> cfg) {
(*cfg)[Sections::kPortRpc].set(Keys::kLimit, std::to_string(limit));
return cfg;
})};
}))};
auto const section = env.app().config().section(Sections::kPortRpc);
// NOLINTBEGIN(bugprone-unchecked-optional-access)
@@ -580,16 +582,27 @@ class ServerStatus_test : public beast::unit_test::Suite, public beast::test::En
BEAST_EXPECT(!ec);
std::vector<std::pair<ip::tcp::socket, boost::beast::multi_buffer>> clients;
int connectionCount{1}; // starts at 1 because the Env already has one
// for JSONRPCCLient
// for nonzero limits, go one past the limit, although failures happen
// at the limit, so this really leads to the last two clients failing.
// for zero limit, pick an arbitrary nonzero number of clients - all
// should connect fine.
// Env owns a persistent JSON-RPC HTTP client connection to port_rpc as
// part of startup, which counts against this port's connection limit.
// This test wants a known starting occupancy of zero, so for nonzero
// limits it deterministically drops that hidden client and waits for
// the server to register the disconnect before opening its own clients.
//
// Starting from zero is important because the port limit rejects once
// the incremented connection count reaches the configured limit. With a
// zero baseline and N = limit + 1 test-owned clients, exactly the last
// two requests should be rejected.
if (limit != 0)
BEAST_EXPECT(env.disconnectClient());
// For nonzero limits, go one past the limit. The port rejects at the
// limit, not only above it, so this yields the last two clients
// failing. For zero limit, pick an arbitrary nonzero number of clients
// and expect them all to succeed.
int const testTo = (limit == 0) ? 50 : limit + 1;
while (connectionCount < testTo)
while (static_cast<int>(clients.size()) < testTo)
{
clients.emplace_back(ip::tcp::socket{ios}, boost::beast::multi_buffer{});
async_connect(clients.back().first, it, yield[ec]);
@@ -597,19 +610,24 @@ class ServerStatus_test : public beast::unit_test::Suite, public beast::test::En
auto req = makeHTTPRequest(ip, port, to_string(jr), {});
async_write(clients.back().first, req, yield[ec]);
BEAST_EXPECT(!ec);
++connectionCount;
}
int readCount = 0;
int successfulReads = 0;
for (auto& [soc, buf] : clients)
{
boost::beast::http::response<boost::beast::http::string_body> resp;
async_read(soc, buf, resp, yield[ec]);
++readCount;
// expect the reads to fail for the clients that connected at or
// above the limit. If limit is 0, all reads should succeed
BEAST_EXPECT((limit == 0 || readCount < limit - 1) ? (!ec) : bool(ec));
if (!ec)
++successfulReads;
}
// This test cares about the exact number of accepted requests, not which
// specific client observed the rejection. With a zero baseline (the
// hidden Env client dropped above), the server accepts until the
// connection count reaches the limit: all clients for limit 0, else
// limit - 1 of the limit + 1 clients (the last two are rejected).
int const expectedReads = (limit == 0) ? static_cast<int>(clients.size()) : limit - 1;
BEAST_EXPECT(successfulReads == expectedReads);
}
void