test: drive the jtx reconnect through a fixture server

The first-request test slept 4 s against the server's 3 s deadline, leaving
1 s for the server's timer to fire on a loaded runner before the request
arrived; it now sleeps 8 s. The keep-alive test compared reconnects() to 0
across both requests, so a legitimate reconnect on a slow first request
failed it; it now checks the count is unchanged across the idle gap.

src/test/jtx/AbstractClient.h

@@ -56,6 +56,13 @@ public:
     /// Get RPC 1.0 or RPC 2.0
     virtual unsigned
     version() const = 0;
+
+    /// Times invoke() replaced a connection the server had closed.
+    virtual unsigned
+    reconnects() const
+    {
+        return 0;
+    }
 };
 
 }  // namespace test

src/test/jtx/Env_test.cpp

@@ -18,6 +18,7 @@
 //==============================================================================
 
 #include <test/jtx.h>
+#include <test/jtx/JSONRPCClient.h>
 #include <xrpld/app/misc/NetworkOPs.h>
 #include <xrpld/app/misc/TxQ.h>
 #include <xrpl/beast/hash/uhash.h>
@@ -27,6 +28,12 @@
 #include <xrpl/protocol/TxFlags.h>
 #include <xrpl/protocol/jss.h>
 
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/ip/tcp.hpp>
+#include <boost/beast/core/flat_buffer.hpp>
+#include <boost/beast/http/read.hpp>
+#include <boost/beast/http/string_body.hpp>
+#include <boost/beast/http/write.hpp>
 #include <boost/lexical_cast.hpp>
 #include <chrono>
 #include <optional>
@@ -917,6 +924,7 @@ public:
 
         Env env{*this};
         env.client().invoke("server_info", {});
+        auto const reconnects = env.client().reconnects();
 
         // The normal loopback message timeout is three seconds. Established
         // Env connections have a separate idle lease because test work can
@@ -924,6 +932,87 @@ public:
         std::this_thread::sleep_for(4s);
         auto const response = env.client().invoke("server_info", {});
         BEAST_EXPECT(response[jss::result][jss::status] == "success");
+        // The server kept the connection; the client did not replace it.
+        BEAST_EXPECT(env.client().reconnects() == reconnects);
+    }
+
+    void
+    testJSONRPCClientSlowFirstRequest()
+    {
+        testcase("JSON-RPC first request survives slow test setup");
+        using namespace std::chrono_literals;
+        using namespace jtx;
+
+        Env env{*this};
+
+        // The client connects when the Env is built, and the server closes a
+        // loopback connection whose first request takes longer than
+        // BaseHTTPPeer::timeoutSecondsLocal (3 s) to arrive. Test setup can
+        // take that long on a loaded machine. Whether or not the server's
+        // timer has fired by now, the request must succeed;
+        // testJSONRPCClientReconnects covers the replacement itself.
+        std::this_thread::sleep_for(4s);
+        auto const response = env.client().invoke("server_info", {});
+        BEAST_EXPECT(response[jss::result][jss::status] == "success");
+    }
+
+    void
+    testJSONRPCClientReconnects()
+    {
+        testcase("JSON-RPC client replaces a connection the server closed");
+        using namespace boost::asio;
+        namespace http = boost::beast::http;
+
+        // A server that closes the client's first connection without
+        // answering, as BaseHTTPPeer does when the first request is late, and
+        // answers on the second.
+        io_context ioc;
+        ip::tcp::acceptor acceptor(ioc, {ip::make_address("127.0.0.1"), 0});
+        ip::tcp::socket first(ioc);
+        ip::tcp::socket second(ioc);
+        boost::beast::flat_buffer buffer;
+        http::request<http::string_body> req;
+        http::response<http::string_body> res{http::status::ok, 11};
+        res.body() = R"({"result":{"status":"success"}})";
+        res.prepare_payload();
+        acceptor.async_accept(first, [&](boost::system::error_code) {
+            boost::system::error_code ignored;
+            first.close(ignored);
+            acceptor.async_accept(second, [&](boost::system::error_code acc) {
+                if (acc)
+                    return;
+                http::async_read(
+                    second,
+                    buffer,
+                    req,
+                    [&](boost::system::error_code rd, std::size_t) {
+                        if (!rd)
+                            http::async_write(
+                                second,
+                                res,
+                                [](boost::system::error_code, std::size_t) {});
+                    });
+            });
+        });
+        std::thread server([&] { ioc.run(); });
+
+        auto cfg = jtx::envconfig();
+        (*cfg)["port_rpc"].set("ip", "127.0.0.1");
+        (*cfg)["port_rpc"].set(
+            "port", std::to_string(acceptor.local_endpoint().port()));
+        auto client = makeJSONRPCClient(*cfg);
+        try
+        {
+            auto const response = client->invoke("server_info", {});
+            BEAST_EXPECT(response[jss::result][jss::status] == "success");
+            BEAST_EXPECT(client->reconnects() == 1);
+        }
+        catch (std::exception const& e)
+        {
+            fail(e.what(), __FILE__, __LINE__);
+        }
+        ioc.stop();
+        server.join();
     }
 
     void
@@ -954,6 +1043,8 @@ public:
         testFeatures(all);
         testExceptionalShutdown();
         testJSONRPCClientKeepAlive();
+        testJSONRPCClientSlowFirstRequest();
+        testJSONRPCClientReconnects();
     }
 };
 

src/test/jtx/impl/JSONRPCClient.cpp

@@ -23,6 +23,7 @@
 #include <xrpl/server/Port.h>
 #include <boost/asio.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>
@@ -79,6 +80,30 @@ class JSONRPCClient : public AbstractClient
     boost::beast::multi_buffer bin_;
     boost::beast::multi_buffer bout_;
     unsigned rpc_version_;
+    unsigned reconnects_ = 0;
+
+    // Errors meaning the server closed the keep-alive connection, rather
+    // than a failed request, so the request can be sent again on a fresh one.
+    static bool
+    droppedConnection(boost::system::error_code const& ec)
+    {
+        namespace error = boost::asio::error;
+        return ec == boost::beast::http::error::end_of_stream ||
+            ec == error::eof || ec == error::connection_reset ||
+            ec == error::connection_aborted || ec == error::broken_pipe ||
+            ec == error::not_connected;
+    }
+
+    // Replace the dropped connection, discarding any bytes buffered from it.
+    void
+    reconnect()
+    {
+        boost::system::error_code ec;
+        stream_.close(ec);
+        bin_.clear();
+        stream_.connect(ep_);
+        ++reconnects_;
+    }
 
 public:
     explicit JSONRPCClient(Config const& cfg, unsigned rpc_version)
@@ -133,10 +158,30 @@ public:
             req.body() = to_string(jr);
         }
         req.prepare_payload();
-        write(stream_, req);
 
+        // The connection is opened at construction and kept alive, but the
+        // server closes it when the first request takes longer than
+        // BaseHTTPPeer::timeoutSecondsLocal to arrive (test setup on a loaded
+        // machine can), or when it idles past timeoutSeconds between requests.
+        // If the server has closed it, reconnect and send the request once
+        // more.
         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;
@@ -153,6 +198,12 @@ public:
     {
         return rpc_version_;
     }
+
+    unsigned
+    reconnects() const override
+    {
+        return reconnects_;
+    }
 };
 
 std::unique_ptr<AbstractClient>