Compare commits

...

2 Commits

Author SHA1 Message Date
Pratik Mankawde
926e07b1ed chore: Brace the multi-line deadline assignment for clang-tidy
readability-braces-around-statements (ShortStatementLines: 2) requires braces once clang-format wraps the assignment onto a second line.
2026-09-22 20:51:53 +01:00
Pratik Mankawde
1dc285ddd4 fix: Missing async deadlines and util::spawn context errors
HTTPClient: arm the request deadline on the normal path (the wait was
registered only when expires_after threw, so no request had a timeout),
reset the per-site error state so host fallback is attempted, and on
expiry close the transport instead of negotiating a TLS shutdown that
waits on the unresponsive peer. Ignore a deadline handler that was
already queued when the timer was re-armed for the next host, treat a
timer wait error like a timeout instead of aborting the process, and
complete on a header read error rather than parsing an incomplete
buffer.

ConnectAttempt: bound the upgrade-response read with the step timer, as
every other step of the attempt already is.

util::spawn: build the strand from the context or executor directly.
get_associated_executor is a handler query and does not instantiate for
an io_context or an executor under Boost 1.91.

Tests: HTTPClient timeout, read-error and host-fallback cases;
util::spawn on an io_context, an executor and a strand, plus exception
propagation.

Fixes #8235
2026-09-22 18:10:14 +01:00
5 changed files with 335 additions and 38 deletions

View File

@@ -58,7 +58,8 @@ inline constexpr auto kPropagateExceptions = [](std::exception_ptr ePtr) {
*
* @tparam Ctx The type of the context/strand
* @tparam F The type of the function to execute
* @param ctx The execution context
* @param ctx An execution context (e.g. `io_context`), an executor, or a
* strand. A strand is used as-is; anything else is wrapped in a new strand.
* @param func The function to execute. Must return `void`
*/
template <typename Ctx, typename F>
@@ -74,7 +75,7 @@ spawn(Ctx&& ctx, F&& func)
else
{
boost::asio::spawn(
boost::asio::make_strand(boost::asio::get_associated_executor(std::forward<Ctx>(ctx))),
boost::asio::make_strand(std::forward<Ctx>(ctx)),
std::forward<F>(func),
impl::kPropagateExceptions);
}

View File

@@ -22,7 +22,6 @@
#include <chrono>
#include <cstddef>
#include <cstdlib>
#include <deque>
#include <functional>
#include <iterator>
@@ -155,6 +154,11 @@ public:
boost::asio::ip::resolver_query_base::numeric_service);
query_ = query;
// Each site starts with a clean error state. shutdown_ records the first
// failure of the current attempt only, so a fallback to the next entry
// in deqSites_ is resolved and connected like a fresh request.
shutdown_.clear();
try
{
deadline_.expires_after(timeout_);
@@ -164,13 +168,16 @@ public:
shutdown_ = e.code();
JLOG(j_.trace()) << "expires_after: " << shutdown_.message();
deadline_.async_wait([self = shared_from_this()](boost::system::error_code const& ec) {
self->handleDeadline(ec);
});
}
if (!shutdown_)
{
// A set expiry fires only if a waiter is registered, so the wait
// is armed whenever expires_after succeeded.
deadline_.async_wait([self = shared_from_this()](boost::system::error_code const& ec) {
self->handleDeadline(ec);
});
JLOG(j_.trace()) << "Resolving: " << deqSites_[0];
resolver_.async_resolve(
@@ -183,9 +190,10 @@ public:
self->handleResolve(ecResult, results);
});
}
if (shutdown_)
else
{
invokeComplete(shutdown_);
}
}
void
@@ -195,41 +203,42 @@ public:
{
// Timer canceled because deadline no longer needed.
JLOG(j_.trace()) << "Deadline cancelled.";
// Aborter is done.
return;
}
else if (ecResult)
{
JLOG(j_.trace()) << "Deadline error: " << deqSites_[0] << ": " << ecResult.message();
// Can't do anything sound.
std::abort();
// A handler that was already queued when httpsNext() re-armed the
// timer for the next site sees an expiry in the future. It belongs to
// the finished attempt and must leave the new one alone. Strictly
// greater: a genuine expiry observed within the same clock tick has
// expiry == now and must still be acted on.
if (deadline_.expiry() > std::chrono::steady_clock::now())
{
JLOG(j_.trace()) << "Stale deadline ignored.";
return;
}
else
JLOG(j_.trace()) << "Deadline: " << (ecResult ? ecResult.message() : "arrived");
// Mark us as shutting down. A wait error ends the attempt exactly as a
// timeout does.
if (!shutdown_)
{
JLOG(j_.trace()) << "Deadline arrived.";
// Mark us as shutting down.
// XXX Use our own error code.
shutdown_ = boost::system::error_code{
boost::system::errc::bad_address, boost::system::system_category()};
// Cancel any resolving.
resolver_.cancel();
// Stop the transaction.
socket_.asyncShutdown([self = shared_from_this()](boost::system::error_code const& ec) {
self->handleShutdown(ec);
});
shutdown_ =
ecResult ? ecResult : boost::system::error_code{boost::asio::error::timed_out};
}
}
void
handleShutdown(boost::system::error_code const& ecResult)
{
if (ecResult)
// Cancel any resolving.
resolver_.cancel();
// Close the transport rather than negotiating a TLS shutdown: the
// negotiation waits on the peer, and the deadline has already
// passed. The pending operation's handler then runs with
// shutdown_ set and reports the timeout.
boost::system::error_code ec;
socket_.lowestLayer().close(ec);
if (ec)
{
JLOG(j_.trace()) << "Shutdown error: " << deqSites_[0] << ": " << ecResult.message();
JLOG(j_.trace()) << "Deadline close error: " << ec.message();
}
}
@@ -366,6 +375,18 @@ public:
void
handleHeader(boost::system::error_code const& ecResult, std::size_t bytesTransferred)
{
// A read error, or a deadline that closed the transport while this read
// was pending, ends the attempt here rather than parsing an incomplete
// header. The deadline records its own code first, so it wins.
if (!shutdown_)
shutdown_ = ecResult;
if (shutdown_)
{
invokeComplete(shutdown_);
return;
}
std::string strHeader{
{std::istreambuf_iterator<char>(&header_)}, std::istreambuf_iterator<char>()};
JLOG(j_.trace()) << "Header: \"" << strHeader << "\"";

View File

@@ -7,6 +7,7 @@
#include <boost/asio/awaitable.hpp>
#include <boost/asio/co_spawn.hpp> // IWYU pragma: keep
#include <boost/asio/detached.hpp>
#include <boost/asio/error.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/socket_base.hpp>
@@ -19,8 +20,10 @@
#include <helpers/TestSink.h>
#include <chrono>
#include <deque>
#include <exception>
#include <map>
#include <optional>
#include <string>
#include <utility>
#include <vector>
@@ -32,6 +35,15 @@ namespace {
// Simple HTTP server using Beast for testing
class TestHTTPServer
{
public:
/**
* What the server does with a connection once it has read the request:
* Reply serves the configured status, headers and body; Stall holds the
* connection open and never replies; CloseAfterRead hangs up without
* replying.
*/
enum class Behaviour { Reply, Stall, CloseAfterRead };
private:
boost::asio::io_context ioc_;
boost::asio::ip::tcp::acceptor acceptor_;
@@ -45,13 +57,28 @@ private:
std::string responseBody_;
unsigned int statusCode_{200};
/**
* How a connection is handled once its request is read. Anything other
* than Reply leaves the client to finish through its own error handling.
*/
Behaviour behaviour_{Behaviour::Reply};
/**
* The socket of the most recent stalled connection. Holding it open sends
* no EOF to the client, so the client's read stays pending until the
* client's own deadline fires and closes the client side.
*/
std::optional<boost::asio::ip::tcp::socket> stalledSocket_;
beast::Journal j_;
public:
TestHTTPServer() : acceptor_(ioc_), j_(TestSink::instance())
{
// Bind to any available port
endpoint_ = {boost::asio::ip::tcp::v4(), 0};
// Bind to a fixed loopback address (rather than 0.0.0.0) so that a
// sibling loopback address such as 127.0.0.2 has no listener, which
// the fallback test relies on.
endpoint_ = {boost::asio::ip::make_address("127.0.0.1"), 0};
acceptor_.open(endpoint_.protocol());
acceptor_.set_option(boost::asio::socket_base::reuse_address(true));
acceptor_.bind(endpoint_);
@@ -103,6 +130,18 @@ public:
statusCode_ = code;
}
/**
* Choose what happens to each connection after its request is read.
*
* @param behaviour Reply to serve the configured response, Stall to hold
* the connection open with no reply, CloseAfterRead to hang up instead.
*/
void
setBehaviour(Behaviour behaviour)
{
behaviour_ = behaviour;
}
void
stop()
{
@@ -155,6 +194,24 @@ private:
co_await boost::beast::http::async_read(
socket, buffer, req, boost::asio::use_awaitable);
if (behaviour_ == Behaviour::Stall)
{
// Hold the connection open and never reply. Moving the socket
// into a member keeps it alive after this coroutine returns,
// so the client's read stays pending until its own deadline
// fires. The accept loop continues and stop() can still end it.
stalledSocket_.emplace(std::move(socket));
co_return;
}
if (behaviour_ == Behaviour::CloseAfterRead)
{
// Hang up without replying, so the client's header read ends
// with EOF and no bytes.
socket.close();
co_return;
}
// Create response
boost::beast::http::response<boost::beast::http::string_body> res;
res.version(req.version());
@@ -366,3 +423,143 @@ TEST_F(HTTPClientTest, different_status_codes)
EXPECT_EQ(resultStatus, static_cast<int>(status));
}
}
TEST_F(HTTPClientTest, request_times_out_on_stalled_peer)
{
// A peer that reads the request but never replies must not leave the
// client waiting: the request deadline fires and completes the handler
// exactly once with timed_out. This depends on the deadline wait being
// armed on the normal, non-throwing expires_after path.
TestHTTPServer server;
server.setBehaviour(TestHTTPServer::Behaviour::Stall);
int completions{0};
int resultStatus{-1};
boost::system::error_code resultError;
HTTPClient::get(
false, // no SSL
server.ioc(),
"127.0.0.1",
server.port(),
"/stall",
1024, // max response size
std::chrono::seconds(1),
[&](boost::system::error_code const& ec, int status, std::string const&) -> bool {
resultError = ec;
resultStatus = status;
++completions;
// Close the acceptor so the accept loop ends and run_for drains.
server.stop();
return false; // don't retry
},
j_);
// Bounded wall-clock drive; the 1s deadline must fire well within this.
server.ioc().run_for(std::chrono::seconds(4));
// Stop unconditionally so the accept loop ends and the fixture's
// finished() check holds even when the client never completes; a
// regression then fails an EXPECT instead of aborting the binary.
server.stop();
server.ioc().poll();
EXPECT_EQ(completions, 1);
EXPECT_EQ(resultError, boost::asio::error::timed_out);
EXPECT_EQ(resultStatus, 0);
EXPECT_TRUE(server.finished());
}
TEST_F(HTTPClientTest, falls_back_to_next_site_after_connect_failure)
{
// When the first site cannot be reached, the client must fall back to the
// next site and report that site's result, not the first site's connect
// error. This depends on shutdown_ being cleared at the start of each
// attempt in httpsNext().
TestHTTPServer server;
std::string const testBody = "fallback body";
server.setResponseBody(testBody);
server.setHeader("Content-Length", std::to_string(testBody.size()));
// First site: a loopback address the server does not listen on. Where the
// whole 127/8 block is local (Linux) the connect is refused at once; where
// 127.0.0.2 is not a configured loopback address (macOS) it is unreachable
// and the 1 s deadline ends the attempt instead. Either way the client
// must move on to the second site, which is the server.
std::deque<std::string> const sites{"127.0.0.2", "127.0.0.1"};
int completions{0};
int resultStatus{-1};
std::string resultData;
boost::system::error_code resultError;
HTTPClient::get(
false, // no SSL
server.ioc(),
sites,
server.port(),
"/fallback",
1024, // max response size
std::chrono::seconds(1),
[&](boost::system::error_code const& ec, int status, std::string const& data) -> bool {
resultError = ec;
resultStatus = status;
resultData = data;
++completions;
server.stop();
return false; // don't retry
},
j_);
// Bounded wall-clock drive: worst case is one 1 s deadline on the first
// site followed by the real exchange on the second.
server.ioc().run_for(std::chrono::seconds(6));
server.stop();
server.ioc().poll();
EXPECT_EQ(completions, 1);
EXPECT_FALSE(resultError);
EXPECT_EQ(resultStatus, 200);
EXPECT_EQ(resultData, testBody);
EXPECT_TRUE(server.finished());
}
TEST_F(HTTPClientTest, reports_read_error_when_peer_closes_without_reply)
{
// A peer that reads the request and hangs up must surface the read error
// itself, not a parse failure of the empty header buffer. This depends on
// handleHeader() recording its own error code before it looks at the
// buffer. The client timeout is longer than the drive window so that a
// timeout cannot stand in for the read error.
TestHTTPServer server;
server.setBehaviour(TestHTTPServer::Behaviour::CloseAfterRead);
int completions{0};
int resultStatus{-1};
boost::system::error_code resultError;
HTTPClient::get(
false, // no SSL
server.ioc(),
"127.0.0.1",
server.port(),
"/hangup",
1024, // max response size
std::chrono::seconds(10),
[&](boost::system::error_code const& ec, int status, std::string const&) -> bool {
resultError = ec;
resultStatus = status;
++completions;
server.stop();
return false; // don't retry
},
j_);
server.ioc().run_for(std::chrono::seconds(4));
server.stop();
server.ioc().poll();
EXPECT_EQ(completions, 1);
EXPECT_EQ(resultError, boost::asio::error::eof);
EXPECT_EQ(resultStatus, 0);
EXPECT_TRUE(server.finished());
}

View File

@@ -0,0 +1,75 @@
#include <xrpl/server/detail/Spawn.h>
#include <boost/asio/io_context.hpp>
#include <boost/asio/spawn.hpp>
#include <boost/asio/strand.hpp>
#include <gtest/gtest.h>
#include <stdexcept>
using namespace xrpl;
namespace {
// util::spawn dispatches on whether its argument is already a strand. In-tree
// callers pass a strand, so these tests are what exercise the non-strand branch
// (which wraps the argument in a new strand): they force it to compile and run
// for a bare io_context and for an io_context executor. A coroutine spawned on
// each context form must execute exactly once while the io_context runs.
TEST(SpawnTest, runs_on_io_context_lvalue)
{
boost::asio::io_context ioc;
int runs = 0;
util::spawn(ioc, [&](boost::asio::yield_context) { ++runs; });
EXPECT_EQ(runs, 0); // deferred until the context runs
ioc.run();
EXPECT_EQ(runs, 1);
}
TEST(SpawnTest, runs_on_executor_lvalue)
{
boost::asio::io_context ioc;
auto executor = ioc.get_executor();
int runs = 0;
util::spawn(executor, [&](boost::asio::yield_context) { ++runs; });
EXPECT_EQ(runs, 0);
ioc.run();
EXPECT_EQ(runs, 1);
}
TEST(SpawnTest, runs_on_strand)
{
boost::asio::io_context ioc;
auto strand = boost::asio::make_strand(ioc);
int runs = 0;
util::spawn(strand, [&](boost::asio::yield_context) { ++runs; });
EXPECT_EQ(runs, 0);
ioc.run();
EXPECT_EQ(runs, 1);
}
TEST(SpawnTest, propagates_exception_to_run)
{
boost::asio::io_context ioc;
util::spawn(
ioc, [](boost::asio::yield_context) { throw std::runtime_error("spawned failure"); });
// kPropagateExceptions must rethrow out of io_context::run(), preserving
// both the type and the message rather than swallowing the exception.
try
{
ioc.run();
FAIL() << "expected the spawned exception to propagate";
}
catch (std::runtime_error const& e)
{
EXPECT_STREQ(e.what(), "spawned failure");
}
}
} // namespace

View File

@@ -307,6 +307,9 @@ ConnectAttempt::onWrite(error_code ec)
return;
}
// The upgrade response is bounded by the same timer as every other step
// of the attempt; onRead() cancels it on entry.
setTimer();
boost::beast::http::async_read(
stream_,
readBuf_,