Test: Make the connection-reuse tests exercise reuse again

libraries/libfc/test/network/test_http_client.cpp

@@ -113,10 +113,15 @@ class scripted_http_server {
    /** Stop the response script and release its worker. */
    ~scripted_http_server() {
       _stop = true;
-      boost::system::error_code ec;
-      _acceptor.close(ec);
-      _socket.cancel(ec);
-      _socket.close(ec);
+      {
+         std::scoped_lock lock(_close_mutex);
+         boost::system::error_code ec;
+         _acceptor.close(ec);
+         // Closing does not wake a handler blocked reading the socket on Linux; shutting it down does.
+         _socket.shutdown(tcp::socket::shutdown_both, ec);
+         _socket.cancel(ec);
+         _socket.close(ec);
+      }
       unblock_accept();
       if (_worker.joinable()) {
          _worker.join();
@@ -126,9 +131,6 @@ class scripted_http_server {
    /** Return the ephemeral loopback port assigned to this server. */
    uint16_t port() const { return _port; }
 
-   /** Return whether the configured server script has completed. */
-   bool finished() const { return _finished.load(); }
-
 private:
    /** Connect to the listener so a platform that does not cancel synchronous accept can exit. */
    void unblock_accept() {
@@ -143,6 +145,7 @@ class scripted_http_server {
       boost::system::error_code ec;
       for (size_t connection_index = 0; connection_index < _connections_to_accept; ++connection_index) {
          if (connection_index != 0) {
+            std::scoped_lock lock(_close_mutex);
             _socket = tcp::socket(_io);
          }
          _acceptor.accept(_socket, ec);
@@ -164,11 +167,12 @@ class scripted_http_server {
          }
          _handler(_socket, _stop);
          if (connection_index + 1 < _connections_to_accept) {
+            std::scoped_lock lock(_close_mutex);
             _socket.close(ec);
          }
       }
+      std::scoped_lock lock(_close_mutex);
       _acceptor.close(ec);
-      _finished = true;
    }
 
    boost::asio::io_context _io;
@@ -179,7 +183,8 @@ class scripted_http_server {
    bool _consume_request_header;
    size_t _connections_to_accept;
    std::atomic_bool _stop{false};
-   std::atomic_bool _finished{false};
+   /// Serializes the listener and socket closes the worker and the destructor both reach; asio's aren't thread-safe.
+   std::mutex _close_mutex;
    std::thread _worker;
 };
 
@@ -394,6 +399,32 @@ std::string keep_alive_metadata_response() {
           "Connection: keep-alive\r\n\r\n{}";
 }
 
+/**
+ * Answer the metadata request on @p socket with keep-alive, then drop the connection unanswered once the next request
+ * arrives on it, recording that arrival in @p request_arrived. The connection stays open until then because the
+ * transport discards, rather than reuses, an idle connection whose peer has already closed it. Dropping only shuts the
+ * socket down, and only once a request arrived; the close is left to the fixture, which serializes it with teardown.
+ */
+void serve_metadata_then_drop_next_request(tcp::socket& socket, std::atomic_bool& request_arrived) {
+   if (!write_bytes(socket, keep_alive_metadata_response()) || read_request_header(socket).empty())
+      return;
+   request_arrived = true;
+   boost::system::error_code ec;
+   socket.shutdown(tcp::socket::shutdown_both, ec);
+}
+
+/**
+ * Answer the request the fixture already read, and every later one on @p socket, with a keep-alive response until the
+ * peer closes it. Keeping the connection open leaves the client's pool, not the transport's closed-peer check, as the
+ * only thing that decides whether it is reused.
+ */
+void answer_keep_alive_until_closed(tcp::socket& socket) {
+   do {
+      if (!write_bytes(socket, keep_alive_metadata_response()))
+         return;
+   } while (!read_request_header(socket).empty());
+}
+
 /** Return the JSON error envelope a remote node emits alongside an HTTP 500 response. */
 std::string remote_error_response_body() {
    fc::mutable_variant_object detail;
@@ -463,6 +494,24 @@ uint16_t unconnectable_loopback_port() {
    return port;
 }
 
+/**
+ * Wait until loopback @p port refuses connections; false if it is still not refusing after 1000 probes, about a second
+ * on loopback. A listener that has just closed keeps accepting for a few milliseconds on the hosts described at
+ * unconnectable_loopback_port.
+ */
+bool wait_until_refused(uint16_t port) {
+   boost::asio::io_context io;
+   for (size_t attempt = 0; attempt < 1'000; ++attempt) {
+      tcp::socket probe(io);
+      boost::system::error_code ec;
+      probe.connect(tcp::endpoint(boost::asio::ip::address_v4::loopback(), port), ec);
+      if (ec == boost::asio::error::connection_refused)
+         return true;
+      std::this_thread::sleep_for(1ms);
+   }
+   return false;
+}
+
 /** Return the URL for @p server. */
 fc::url server_url(const scripted_http_server& server) {
    return fc::url("http://127.0.0.1:" + std::to_string(server.port()) + "/download");
@@ -786,9 +835,7 @@ BOOST_AUTO_TEST_CASE(idle_connection_pool_cap_can_disable_reuse) {
    scripted_http_server server(
       [&](tcp::socket& socket, const std::atomic_bool&) {
          ++connections;
-         (void)write_bytes(socket, "HTTP/1.1 200 OK\r\n"
-                                   "Content-Length: 2\r\n"
-                                   "Connection: keep-alive\r\n\r\n{}");
+         answer_keep_alive_until_closed(socket);
       },
       true, 2);
    fc::http::transport transport(fc::http::transport_options{
@@ -940,9 +987,7 @@ BOOST_AUTO_TEST_CASE(expired_idle_connection_is_not_reused) {
    scripted_http_server server(
       [&](tcp::socket& socket, const std::atomic_bool&) {
          ++connections;
-         (void)write_bytes(socket, "HTTP/1.1 200 OK\r\n"
-                                   "Content-Length: 2\r\n"
-                                   "Connection: keep-alive\r\n\r\n{}");
+         answer_keep_alive_until_closed(socket);
       },
       true, 2);
    fc::http::transport transport(fc::http::transport_options{
@@ -2286,16 +2331,14 @@ BOOST_AUTO_TEST_CASE(healthy_metadata_connection_is_reused_for_download) {
    BOOST_CHECK_EQUAL(read_file(output), exact_body);
 }
 
-/// A cached connection closed after metadata should retry the idempotent download once.
+/// A cached connection that goes stale in use should retry the idempotent download once on a fresh connection.
 BOOST_AUTO_TEST_CASE(stale_metadata_connection_retries_download_on_fresh_connection) {
    std::atomic_size_t connection_index{0};
+   std::atomic_bool reused_request_received{false};
    scripted_http_server server(
       [&](tcp::socket& socket, const std::atomic_bool&) {
          if (connection_index.fetch_add(1) == 0) {
-            if (write_bytes(socket, keep_alive_metadata_response())) {
-               boost::system::error_code ec;
-               socket.shutdown(tcp::socket::shutdown_both, ec);
-            }
+            serve_metadata_then_drop_next_request(socket, reused_request_received);
             return;
          }
          write_bytes(socket, fixed_length_header(exact_body_bytes) + std::string(exact_body));
@@ -2313,17 +2356,16 @@ BOOST_AUTO_TEST_CASE(stale_metadata_connection_retries_download_on_fresh_connect
    options.retry_failed_reused_connection = true;
    BOOST_REQUIRE_NO_THROW(
       client.post_to_file(server_url(server), fc::variant(fc::mutable_variant_object()), output, options));
+   BOOST_CHECK(reused_request_received);
    BOOST_CHECK_EQUAL(connection_index.load(), 2U);
    BOOST_CHECK_EQUAL(read_file(output), exact_body);
 }
 
-/// A failed reconnect after stale reuse must unwind without touching an invalid connection iterator.
+/// A failed reconnect after a cached connection goes stale in use surfaces the connect failure and leaves no files.
 BOOST_AUTO_TEST_CASE(stale_metadata_reconnect_failure_cleans_up_safely) {
-   scripted_http_server server([](tcp::socket& socket, const std::atomic_bool&) {
-      if (write_bytes(socket, keep_alive_metadata_response())) {
-         boost::system::error_code ec;
-         socket.shutdown(tcp::socket::shutdown_both, ec);
-      }
+   std::atomic_bool reused_request_received{false};
+   scripted_http_server server([&](tcp::socket& socket, const std::atomic_bool&) {
+      serve_metadata_then_drop_next_request(socket, reused_request_received);
    });
    fc::temp_directory temp;
    const auto output = temp.path() / "failed-reconnect.bin";
@@ -2333,17 +2375,25 @@ BOOST_AUTO_TEST_CASE(stale_metadata_reconnect_failure_cleans_up_safely) {
 
    BOOST_REQUIRE_NO_THROW(
       client.post_sync(server_url(server), fc::variant(fc::mutable_variant_object()), metadata_deadline));
-   for (size_t wait_count = 0; wait_count < 1'000 && !server.finished(); ++wait_count) {
-      std::this_thread::sleep_for(1ms);
-   }
-   BOOST_REQUIRE(server.finished());
    auto options = download_options(exact_body_bytes);
    options.retry_failed_reused_connection = true;
+   // The server stops listening when it drops the stale connection; hold the retry until that port refuses.
+   size_t connecting_phases = 0;
+   bool refused_before_reconnect = false;
+   options.status_callback = [&](const fc::http_file_download_status& status) {
+      if (status.phase == fc::http_file_download_phase::connecting && ++connecting_phases == 2)
+         refused_before_reconnect = wait_until_refused(server.port());
+   };
    BOOST_CHECK_EXCEPTION(
       client.post_to_file(server_url(server), fc::variant(fc::mutable_variant_object()), output, options),
       fc::exception, [](const fc::exception& error) {
-         return error.to_detail_string().find("Failed to connect") != std::string::npos;
+         const auto detail = error.to_detail_string();
+         return detail.find("Failed to connect") != std::string::npos &&
+                detail.find("retry_exhausted") == std::string::npos;
       });
+   BOOST_CHECK(reused_request_received);
+   BOOST_CHECK_EQUAL(connecting_phases, 2U);
+   BOOST_CHECK(refused_before_reconnect);
    check_download_files_removed(output);
 }
 
@@ -2465,7 +2515,6 @@ BOOST_AUTO_TEST_CASE(truncated_fixed_length_response_is_rejected_and_removed) {
       write_bytes(socket, fixed_length_header(exact_body_bytes) + "short");
       boost::system::error_code ec;
       socket.shutdown(tcp::socket::shutdown_send, ec);
-      socket.close(ec);
    });
    fc::temp_directory temp;
    const auto output = temp.path() / "truncated-fixed.bin";

libraries/libfc/test/network/test_json_rpc_client.cpp

@@ -184,8 +184,8 @@ class fixed_response_http_server {
 };
 
 /**
- * JSON-RPC endpoint that either serves two calls on one connection or closes
- * the first keep-alive connection before accepting the second call.
+ * JSON-RPC endpoint that either serves two calls on one connection, or drops the first keep-alive connection when the
+ * second call arrives on it and serves that call's replay on a new connection.
  */
 class reusable_json_rpc_server {
 public:
@@ -244,22 +244,30 @@ class reusable_json_rpc_server {
       return socket;
    }
 
-   /** Read one complete request and send the matching JSON-RPC response. */
-   bool serve_request(tcp::socket& socket, int64_t response_id, std::string_view result, bool keep_alive) {
+   /** Read and count one complete request; false if the read failed. */
+   bool read_request(tcp::socket& socket) {
       boost::beast::flat_buffer request_buffer;
       boost::beast::http::request<boost::beast::http::string_body> request;
       boost::system::error_code error;
       boost::beast::http::read(socket, request_buffer, request, error);
       if (error)
          return false;
       _request_count.fetch_add(1);
+      return true;
+   }
+
+   /** Read one complete request and send the matching JSON-RPC response. */
+   bool serve_request(tcp::socket& socket, int64_t response_id, std::string_view result, bool keep_alive) {
+      if (!read_request(socket))
+         return false;
 
       boost::beast::http::response<boost::beast::http::string_body> response{boost::beast::http::status::ok, 11};
       response.set(boost::beast::http::field::content_type, "application/json");
       response.keep_alive(keep_alive);
       response.body() =
          "{\"jsonrpc\":\"2.0\",\"id\":" + std::to_string(response_id) + ",\"result\":\"" + std::string(result) + "\"}";
       response.prepare_payload();
+      boost::system::error_code error;
       boost::beast::http::write(socket, response, error);
       return !error;
    }
@@ -277,6 +285,9 @@ class reusable_json_rpc_server {
          return;
       }
 
+      // Hold the connection open until the second call arrives on it, so the client really reuses it, then drop that
+      // call unanswered.
+      (void)read_request(*first);
       boost::system::error_code error;
       first->shutdown(tcp::socket::shutdown_both, error);
       first->close(error);
@@ -527,7 +538,8 @@ BOOST_AUTO_TEST_CASE(idempotent_call_recovers_from_a_stale_cached_connection) {
    BOOST_CHECK_EQUAL(client.call_idempotent("wire_first_probe").as_string(), "first");
    BOOST_CHECK_EQUAL(client.call_idempotent("wire_second_probe").as_string(), "second");
    BOOST_CHECK_EQUAL(server.connection_count(), 2U);
-   BOOST_CHECK_EQUAL(server.request_count(), 2U);
+   // The second call reaches the server twice: on the stale connection, which drops it, and as the replay.
+   BOOST_CHECK_EQUAL(server.request_count(), 3U);
 }
 
 /// Caller-supplied retry options cannot make a default call replay.