mirror of
https://github.com/XRPLF/rippled.git
synced 2026-09-24 06:00:17 +00:00
Compare commits
2 Commits
copilot/ad
...
pratik/net
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
926e07b1ed | ||
|
|
1dc285ddd4 |
@@ -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);
|
||||
}
|
||||
|
||||
@@ -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 << "\"";
|
||||
|
||||
@@ -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());
|
||||
}
|
||||
|
||||
75
src/tests/libxrpl/server/Spawn.cpp
Normal file
75
src/tests/libxrpl/server/Spawn.cpp
Normal 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
|
||||
@@ -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_,
|
||||
|
||||
Reference in New Issue
Block a user