Merge branch 'develop' into pratik/net-async-deadlines-and-spawn-ctx

readability-braces-around-statements (ShortStatementLines: 2) requires braces once clang-format wraps the assignment onto a second line.

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 <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);
     }

src/libxrpl/net/HTTPClient.cpp

@@ -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);
                 });
         }
-
+        else
-        if (shutdown_)
+        {
             invokeComplete(shutdown_);
+        }
     }
 
     void
@@ -195,41 +203,42 @@ public:
         {
             // Timer canceled because deadline no longer needed.
             JLOG(j_.trace()) << "Deadline cancelled.";
-
+            return;
-            // Aborter is done.
         }
-        else if (ecResult)
-        {
-            JLOG(j_.trace()) << "Deadline error: " << deqSites_[0] << ": " << ecResult.message();
 
-            // Can't do anything sound.
+        // A handler that was already queued when httpsNext() re-armed the
-            std::abort();
+        // 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.";
+            shutdown_ =
-
+                ecResult ? ecResult : boost::system::error_code{boost::asio::error::timed_out};
-            // 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
+        // Cancel any resolving.
-    handleShutdown(boost::system::error_code const& ecResult)
+        resolver_.cancel();
-    {
+
-        if (ecResult)
+        // 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 << "\"";

src/tests/libxrpl/net/HTTPClient.cpp

@@ -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
+        // Bind to a fixed loopback address (rather than 0.0.0.0) so that a
-        endpoint_ = {boost::asio::ip::tcp::v4(), 0};
+        // 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());
+}

src/tests/libxrpl/server/Spawn.cpp

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

src/xrpld/overlay/detail/ConnectAttempt.cpp

@@ -307,6 +307,9 @@ ConnectAttempt::onWrite(ErrorCode 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_,