diff --git a/libraries/libfc/test/network/test_http_client.cpp b/libraries/libfc/test/network/test_http_client.cpp index 005e9fe535..fc104f7f51 100644 --- a/libraries/libfc/test/network/test_http_client.cpp +++ b/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"; diff --git a/libraries/libfc/test/network/test_json_rpc_client.cpp b/libraries/libfc/test/network/test_json_rpc_client.cpp index 05ad132013..9b7f7cf416 100644 --- a/libraries/libfc/test/network/test_json_rpc_client.cpp +++ b/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,8 +244,8 @@ 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 request; boost::system::error_code error; @@ -253,6 +253,13 @@ class reusable_json_rpc_server { 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 response{boost::beast::http::status::ok, 11}; response.set(boost::beast::http::field::content_type, "application/json"); @@ -260,6 +267,7 @@ class reusable_json_rpc_server { 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.