From 4650478a590b956510a15e2fe108a9234fa47f53 Mon Sep 17 00:00:00 2001 From: Coldwings Date: Wed, 22 Jul 2026 07:10:42 +0800 Subject: [PATCH] test(http): synchronize client cancellation regressions --- include/elio/http/client_base.hpp | 16 +++ include/elio/http/http_client.hpp | 3 + include/elio/http/sse_client.hpp | 3 + include/elio/http/websocket_client.hpp | 3 + include/elio/io/io_awaitables.hpp | 23 ++++ tests/unit/test_http_client.cpp | 160 ++++++++++++++++++++----- 6 files changed, 181 insertions(+), 27 deletions(-) diff --git a/include/elio/http/client_base.hpp b/include/elio/http/client_base.hpp index 3ca69e71..5f253b58 100644 --- a/include/elio/http/client_base.hpp +++ b/include/elio/http/client_base.hpp @@ -35,6 +35,22 @@ namespace elio::http { namespace detail { +#ifdef ELIO_RUNTIME_TEST_HOOKS +// Allows cancellation regression tests to wait until a client response recv +// has been staged. Keeping the hook here gives HTTP, WebSocket, and SSE the +// same synchronization contract. +inline std::atomic observe_client_response_read_entry_for_test{false}; +inline std::atomic client_response_read_staged_for_test{false}; + +inline void arm_client_response_read_observer_for_test() noexcept { + if (observe_client_response_read_entry_for_test.load( + std::memory_order_acquire)) { + io::detail::arm_next_cancellable_recv_staged_for_test( + client_response_read_staged_for_test); + } +} +#endif + inline size_t next_rotation_offset(const std::string& host, uint16_t port, size_t count) { if (count == 0) { return 0; diff --git a/include/elio/http/http_client.hpp b/include/elio/http/http_client.hpp index a40c2854..478d5e83 100644 --- a/include/elio/http/http_client.hpp +++ b/include/elio/http/http_client.hpp @@ -540,6 +540,9 @@ class client { pending_response_bytes.clear(); } else { io::io_result read_result{}; +#ifdef ELIO_RUNTIME_TEST_HOOKS + detail::arm_client_response_read_observer_for_test(); +#endif std::shared_ptr> read_timed_out; if (deadline_enforced) { // Use remaining time from the absolute response deadline diff --git a/include/elio/http/sse_client.hpp b/include/elio/http/sse_client.hpp index c53c532b..d3e999e0 100644 --- a/include/elio/http/sse_client.hpp +++ b/include/elio/http/sse_client.hpp @@ -628,6 +628,9 @@ class sse_client { } io::io_result read_result{}; +#ifdef ELIO_RUNTIME_TEST_HOOKS + http::detail::arm_client_response_read_observer_for_test(); +#endif if (deadline_enforced) { auto remaining = response_deadline - std::chrono::steady_clock::now(); diff --git a/include/elio/http/websocket_client.hpp b/include/elio/http/websocket_client.hpp index 8f3e6775..7bc1c595 100644 --- a/include/elio/http/websocket_client.hpp +++ b/include/elio/http/websocket_client.hpp @@ -476,6 +476,9 @@ class ws_client { } io::io_result read_result{}; +#ifdef ELIO_RUNTIME_TEST_HOOKS + http::detail::arm_client_response_read_observer_for_test(); +#endif if (deadline_enforced) { auto remaining = response_deadline - std::chrono::steady_clock::now(); diff --git a/include/elio/io/io_awaitables.hpp b/include/elio/io/io_awaitables.hpp index 53df6403..5870fe1c 100644 --- a/include/elio/io/io_awaitables.hpp +++ b/include/elio/io/io_awaitables.hpp @@ -33,6 +33,25 @@ inline io_context& current_io_context() noexcept { namespace detail { +#ifdef ELIO_RUNTIME_TEST_HOOKS +inline thread_local std::atomic* + next_cancellable_recv_staged_for_test = nullptr; + +inline void arm_next_cancellable_recv_staged_for_test( + std::atomic& staged) noexcept { + next_cancellable_recv_staged_for_test = &staged; +} + +inline void publish_cancellable_recv_staged_for_test() noexcept { + auto* staged = next_cancellable_recv_staged_for_test; + next_cancellable_recv_staged_for_test = nullptr; + if (staged != nullptr) { + staged->store(true, std::memory_order_release); + staged->notify_all(); + } +} +#endif + inline constexpr int socket_no_sigpipe_flag = #ifdef MSG_NOSIGNAL MSG_NOSIGNAL; @@ -1339,6 +1358,10 @@ class cancellable_async_recv_awaitable : public io_awaitable_base { return; } +#ifdef ELIO_RUNTIME_TEST_HOOKS + detail::publish_cancellable_recv_staged_for_test(); +#endif + // Post-registration race: cancel may have fired between on_cancel() // and setting state->op above. Re-check after prepare so the backend // can actually find the staged operation and abort it. diff --git a/tests/unit/test_http_client.cpp b/tests/unit/test_http_client.cpp index b73ea992..0fc99d2d 100644 --- a/tests/unit/test_http_client.cpp +++ b/tests/unit/test_http_client.cpp @@ -19,6 +19,7 @@ #include #include #include +#include #include #include #if defined(ELIO_HAS_TLS) && ELIO_HAS_TLS @@ -55,6 +56,39 @@ constexpr bool running_under_tsan() { #endif } +struct client_response_read_observer_guard { + client_response_read_observer_guard() { + elio::http::detail::client_response_read_staged_for_test.store( + false, std::memory_order_release); + elio::http::detail::observe_client_response_read_entry_for_test.store( + true, std::memory_order_release); + } + + ~client_response_read_observer_guard() { + elio::http::detail::observe_client_response_read_entry_for_test.store( + false, std::memory_order_release); + } +}; + +bool wait_for_client_response_read_staged() { + for (int i = 0; i < 500; ++i) { + if (elio::http::detail::client_response_read_staged_for_test.load( + std::memory_order_acquire)) { + return true; + } + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + return elio::http::detail::client_response_read_staged_for_test.load( + std::memory_order_acquire); +} + +bool wait_for_flag(const std::atomic& flag) { + for (int i = 0; i < 500 && !flag.load(std::memory_order_acquire); ++i) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + return flag.load(std::memory_order_acquire); +} + std::string make_url(uint16_t port, std::string_view path = "/") { return std::string("http://127.0.0.1:") + std::to_string(port) + std::string(path); @@ -88,6 +122,18 @@ task read_request_headers(elio::net::tcp_stream& s) { co_return accum; } +task read_request_headers(elio::net::tcp_stream& s, + elio::coro::cancel_token token) { + std::string accum; + char buf[1024]; + while (accum.find("\r\n\r\n") == std::string::npos) { + auto r = co_await s.read(buf, sizeof(buf), token); + if (r.result <= 0) co_return accum; + accum.append(buf, static_cast(r.result)); + } + co_return accum; +} + task drain_request_headers(elio::net::tcp_stream& s) { (void)co_await read_request_headers(s); co_return; @@ -1056,6 +1102,8 @@ TEST_CASE("HTTP client cancellation aborts a pending response read", SKIP("connect-cancellation fake server drain is covered by normal and ASAN runs"); } + client_response_read_observer_guard response_read_guard; + auto listener = tcp_listener::bind(ipv4_address("127.0.0.1", 0)); REQUIRE(listener.has_value()); uint16_t port = listener->local_address().port(); @@ -1065,25 +1113,34 @@ TEST_CASE("HTTP client cancellation aborts a pending response read", std::atomic client_done{false}; std::atomic client_failed{false}; + std::atomic server_done{false}; + std::atomic server_received_complete_headers{false}; std::atomic client_errno{0}; std::atomic client_elapsed_ms{-1}; elio::coro::cancel_source cancel_source; + elio::coro::cancel_source server_cancel_source; sched.go([&]() -> task { - auto stream = co_await listener->accept(); - REQUIRE(stream.has_value()); - co_await drain_request_headers(*stream); + co_await elio::set_affinity(0); + auto stream = co_await listener->accept(server_cancel_source.get_token()); + if (!stream) { + server_done.store(true, std::memory_order_release); + co_return; + } + const auto headers = co_await read_request_headers( + *stream, server_cancel_source.get_token()); + server_received_complete_headers.store( + headers.find("\r\n\r\n") != std::string::npos, + std::memory_order_release); char extra[1]; - (void)co_await stream->read(extra, sizeof(extra)); + (void)co_await stream->read(extra, sizeof(extra), + server_cancel_source.get_token()); stream->shutdown_socket(); + server_done.store(true, std::memory_order_release); }); sched.go([&]() -> task { - co_await elio::time::sleep_for(std::chrono::milliseconds(100)); - cancel_source.cancel(); - }); - - sched.go([&]() -> task { + co_await elio::set_affinity(1); elio::http::client_config cfg; cfg.read_timeout = std::chrono::seconds(0); elio::http::client c(cfg); @@ -1101,6 +1158,8 @@ TEST_CASE("HTTP client cancellation aborts a pending response read", client_done = true; }); + const bool response_read_staged = wait_for_client_response_read_staged(); + cancel_source.cancel(); for (int i = 0; i < 500 && !client_done; ++i) { std::this_thread::sleep_for(std::chrono::milliseconds(10)); } @@ -1110,8 +1169,15 @@ TEST_CASE("HTTP client cancellation aborts a pending response read", std::this_thread::sleep_for(std::chrono::milliseconds(10)); } } + const bool server_received_headers = + wait_for_flag(server_received_complete_headers); + server_cancel_source.cancel(); + const bool server_finished = wait_for_flag(server_done); REQUIRE(sched.shutdown(std::chrono::seconds(10))); + REQUIRE(response_read_staged); + REQUIRE(server_received_headers); + REQUIRE(server_finished); REQUIRE(client_done); REQUIRE(client_failed); REQUIRE(client_errno == ECANCELED); @@ -1125,6 +1191,8 @@ TEST_CASE("WebSocket client cancellation aborts stalled handshake response", SKIP("connect-cancellation fake server drain is covered by normal and ASAN runs"); } + client_response_read_observer_guard response_read_guard; + auto listener = tcp_listener::bind(ipv4_address("127.0.0.1", 0)); REQUIRE(listener.has_value()); uint16_t port = listener->local_address().port(); @@ -1134,25 +1202,34 @@ TEST_CASE("WebSocket client cancellation aborts stalled handshake response", std::atomic client_done{false}; std::atomic client_failed{false}; + std::atomic server_done{false}; + std::atomic server_received_complete_headers{false}; std::atomic client_errno{0}; std::atomic client_elapsed_ms{-1}; elio::coro::cancel_source cancel_source; + elio::coro::cancel_source server_cancel_source; sched.go([&]() -> task { - auto stream = co_await listener->accept(); - REQUIRE(stream.has_value()); - co_await drain_request_headers(*stream); + co_await elio::set_affinity(0); + auto stream = co_await listener->accept(server_cancel_source.get_token()); + if (!stream) { + server_done.store(true, std::memory_order_release); + co_return; + } + const auto headers = co_await read_request_headers( + *stream, server_cancel_source.get_token()); + server_received_complete_headers.store( + headers.find("\r\n\r\n") != std::string::npos, + std::memory_order_release); char extra[1]; - (void)co_await stream->read(extra, sizeof(extra)); + (void)co_await stream->read(extra, sizeof(extra), + server_cancel_source.get_token()); stream->shutdown_socket(); + server_done.store(true, std::memory_order_release); }); sched.go([&]() -> task { - co_await elio::time::sleep_for(std::chrono::milliseconds(100)); - cancel_source.cancel(); - }); - - sched.go([&]() -> task { + co_await elio::set_affinity(1); elio::http::websocket::client_config cfg; cfg.read_timeout = std::chrono::seconds(0); elio::http::websocket::ws_client client(cfg); @@ -1170,6 +1247,8 @@ TEST_CASE("WebSocket client cancellation aborts stalled handshake response", client_done = true; }); + const bool response_read_staged = wait_for_client_response_read_staged(); + cancel_source.cancel(); for (int i = 0; i < 500 && !client_done; ++i) { std::this_thread::sleep_for(std::chrono::milliseconds(10)); } @@ -1179,8 +1258,15 @@ TEST_CASE("WebSocket client cancellation aborts stalled handshake response", std::this_thread::sleep_for(std::chrono::milliseconds(10)); } } + const bool server_received_headers = + wait_for_flag(server_received_complete_headers); + server_cancel_source.cancel(); + const bool server_finished = wait_for_flag(server_done); REQUIRE(sched.shutdown(std::chrono::seconds(10))); + REQUIRE(response_read_staged); + REQUIRE(server_received_headers); + REQUIRE(server_finished); REQUIRE(client_done); REQUIRE(client_failed); REQUIRE(client_errno == ECANCELED); @@ -1194,6 +1280,8 @@ TEST_CASE("SSE client cancellation aborts stalled response headers", SKIP("connect-cancellation fake server drain is covered by normal and ASAN runs"); } + client_response_read_observer_guard response_read_guard; + auto listener = tcp_listener::bind(ipv4_address("127.0.0.1", 0)); REQUIRE(listener.has_value()); uint16_t port = listener->local_address().port(); @@ -1203,25 +1291,34 @@ TEST_CASE("SSE client cancellation aborts stalled response headers", std::atomic client_done{false}; std::atomic client_failed{false}; + std::atomic server_done{false}; + std::atomic server_received_complete_headers{false}; std::atomic client_errno{0}; std::atomic client_elapsed_ms{-1}; elio::coro::cancel_source cancel_source; + elio::coro::cancel_source server_cancel_source; sched.go([&]() -> task { - auto stream = co_await listener->accept(); - REQUIRE(stream.has_value()); - co_await drain_request_headers(*stream); + co_await elio::set_affinity(0); + auto stream = co_await listener->accept(server_cancel_source.get_token()); + if (!stream) { + server_done.store(true, std::memory_order_release); + co_return; + } + const auto headers = co_await read_request_headers( + *stream, server_cancel_source.get_token()); + server_received_complete_headers.store( + headers.find("\r\n\r\n") != std::string::npos, + std::memory_order_release); char extra[1]; - (void)co_await stream->read(extra, sizeof(extra)); + (void)co_await stream->read(extra, sizeof(extra), + server_cancel_source.get_token()); stream->shutdown_socket(); + server_done.store(true, std::memory_order_release); }); sched.go([&]() -> task { - co_await elio::time::sleep_for(std::chrono::milliseconds(100)); - cancel_source.cancel(); - }); - - sched.go([&]() -> task { + co_await elio::set_affinity(1); elio::http::sse::client_config cfg; cfg.auto_reconnect = false; cfg.read_timeout = std::chrono::seconds(0); @@ -1241,6 +1338,8 @@ TEST_CASE("SSE client cancellation aborts stalled response headers", client_done = true; }); + const bool response_read_staged = wait_for_client_response_read_staged(); + cancel_source.cancel(); for (int i = 0; i < 500 && !client_done; ++i) { std::this_thread::sleep_for(std::chrono::milliseconds(10)); } @@ -1250,8 +1349,15 @@ TEST_CASE("SSE client cancellation aborts stalled response headers", std::this_thread::sleep_for(std::chrono::milliseconds(10)); } } + const bool server_received_headers = + wait_for_flag(server_received_complete_headers); + server_cancel_source.cancel(); + const bool server_finished = wait_for_flag(server_done); REQUIRE(sched.shutdown(std::chrono::seconds(10))); + REQUIRE(response_read_staged); + REQUIRE(server_received_headers); + REQUIRE(server_finished); REQUIRE(client_done); REQUIRE(client_failed); REQUIRE(client_errno == ECANCELED);