diff --git a/include/xrpl/server/detail/Spawn.h b/include/xrpl/server/detail/Spawn.h index 1ba32c2f0b..e3faa41106 100644 --- a/include/xrpl/server/detail/Spawn.h +++ b/include/xrpl/server/detail/Spawn.h @@ -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 @@ -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))), + boost::asio::make_strand(std::forward(ctx)), std::forward(func), impl::kPropagateExceptions); } diff --git a/src/libxrpl/net/HTTPClient.cpp b/src/libxrpl/net/HTTPClient.cpp index 5edd8b5b47..7de7d68f81 100644 --- a/src/libxrpl/net/HTTPClient.cpp +++ b/src/libxrpl/net/HTTPClient.cpp @@ -22,7 +22,6 @@ #include #include -#include #include #include #include @@ -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,40 @@ 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_) + shutdown_ = + ecResult ? ecResult : boost::system::error_code{boost::asio::error::timed_out}; + + // 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()) << "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); - }); - } - } - - void - handleShutdown(boost::system::error_code const& ecResult) - { - if (ecResult) - { - JLOG(j_.trace()) << "Shutdown error: " << deqSites_[0] << ": " << ecResult.message(); + JLOG(j_.trace()) << "Deadline close error: " << ec.message(); } } @@ -366,6 +373,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(&header_)}, std::istreambuf_iterator()}; JLOG(j_.trace()) << "Header: \"" << strHeader << "\""; diff --git a/src/tests/libxrpl/net/HTTPClient.cpp b/src/tests/libxrpl/net/HTTPClient.cpp index 01de5ff999..61578851ea 100644 --- a/src/tests/libxrpl/net/HTTPClient.cpp +++ b/src/tests/libxrpl/net/HTTPClient.cpp @@ -7,6 +7,7 @@ #include #include // IWYU pragma: keep #include +#include #include #include #include @@ -19,8 +20,10 @@ #include #include +#include #include #include +#include #include #include #include @@ -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 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 res; res.version(req.version()); @@ -366,3 +423,143 @@ TEST_F(HTTPClientTest, different_status_codes) EXPECT_EQ(resultStatus, static_cast(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 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()); +} diff --git a/src/tests/libxrpl/server/Spawn.cpp b/src/tests/libxrpl/server/Spawn.cpp new file mode 100644 index 0000000000..72863b66a3 --- /dev/null +++ b/src/tests/libxrpl/server/Spawn.cpp @@ -0,0 +1,75 @@ +#include + +#include +#include +#include + +#include + +#include + +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 diff --git a/src/xrpld/overlay/detail/ConnectAttempt.cpp b/src/xrpld/overlay/detail/ConnectAttempt.cpp index b78b8eb7b8..aabebd9b22 100644 --- a/src/xrpld/overlay/detail/ConnectAttempt.cpp +++ b/src/xrpld/overlay/detail/ConnectAttempt.cpp @@ -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_,