From 0b879bb2a7a41e015e49798146f607ecdcc32634 Mon Sep 17 00:00:00 2001 From: Coldwings Date: Sat, 11 Jul 2026 10:21:31 +0800 Subject: [PATCH 1/2] Bound client protocol handshakes by read_timeout --- include/elio/http/client_base.hpp | 28 ++++++ include/elio/http/sse_client.hpp | 53 ++++++++++- include/elio/http/websocket_client.hpp | 55 ++++++++++- include/elio/net/stream.hpp | 15 +++ tests/unit/test_http_client.cpp | 126 ++++++++++++++++++++++++- wiki/API-Reference.md | 6 ++ wiki/WebSocket-SSE.md | 10 ++ 7 files changed, 286 insertions(+), 7 deletions(-) diff --git a/include/elio/http/client_base.hpp b/include/elio/http/client_base.hpp index 3e28c194..93276fb1 100644 --- a/include/elio/http/client_base.hpp +++ b/include/elio/http/client_base.hpp @@ -18,6 +18,8 @@ #include #include +#include + #include #include #include @@ -25,6 +27,7 @@ #include #include #include +#include namespace elio::http { @@ -46,6 +49,31 @@ inline size_t next_rotation_offset(const std::string& host, uint16_t port, size_ return offset; } +/// Spawn a watchdog that shutdown(2)s `fd` after `timeout` elapses. +/// +/// The returned join_handle must be awaited after the I/O operation completes; +/// the caller cancels `watchdog_token` to wake the watchdog early on success. +/// `timed_out` is set only when the deadline fired before cancellation. +inline coro::join_handle +arm_fd_shutdown_watchdog(runtime::scheduler* sched, + int fd, + std::chrono::nanoseconds timeout, + coro::cancel_token watchdog_token, + std::shared_ptr> timed_out) { + return sched->go_joinable( + [fd, timeout, tok = std::move(watchdog_token), + flag = std::move(timed_out)]() -> coro::task { + auto r = co_await elio::time::sleep_for(timeout, tok); + if (r == coro::cancel_result::completed) { + flag->store(true, std::memory_order_release); + if (fd >= 0) { + ::shutdown(fd, SHUT_RDWR); + } + } + co_return; + }); +} + } // namespace detail /// Base configuration shared by all HTTP-based clients diff --git a/include/elio/http/sse_client.hpp b/include/elio/http/sse_client.hpp index 21bfa7bd..2b3ae5b8 100644 --- a/include/elio/http/sse_client.hpp +++ b/include/elio/http/sse_client.hpp @@ -22,6 +22,7 @@ #include #include +#include #include #include #include @@ -543,11 +544,57 @@ class sse_client { // delimiter directly to the SSE event parser. std::string response_data; response_data.reserve(1024); + auto* sched = runtime::scheduler::current(); + const bool deadline_enforced = + sched != nullptr && config_.read_timeout.count() > 0; + const auto response_deadline = + std::chrono::steady_clock::now() + config_.read_timeout; while (true) { - auto read_result = co_await read(buffer_.data(), buffer_.size()); + if (token_.is_cancelled()) { + errno = ECANCELED; + stream_.disconnect(); + state_ = client_state::disconnected; + co_return false; + } + + io::io_result read_result{}; + if (deadline_enforced) { + auto remaining = + response_deadline - std::chrono::steady_clock::now(); + if (remaining.count() <= 0) { + ELIO_LOG_ERROR("SSE response headers timed out after {}s", + config_.read_timeout.count()); + errno = ETIMEDOUT; + stream_.disconnect(); + state_ = client_state::disconnected; + co_return false; + } + + auto timed_out = std::make_shared>(false); + coro::cancel_source watchdog_cancel; + auto watchdog = http::detail::arm_fd_shutdown_watchdog( + sched, stream_.fd(), remaining, + watchdog_cancel.get_token(), timed_out); + read_result = co_await read(buffer_.data(), buffer_.size()); + watchdog_cancel.cancel(); + co_await watchdog; + if (timed_out->load(std::memory_order_acquire)) { + stream_.mark_externally_shut_down(); + ELIO_LOG_ERROR("SSE response headers timed out after {}s", + config_.read_timeout.count()); + errno = ETIMEDOUT; + stream_.disconnect(); + state_ = client_state::disconnected; + co_return false; + } + } else { + read_result = co_await read(buffer_.data(), buffer_.size()); + } + if (read_result.result <= 0) { ELIO_LOG_ERROR("Failed to read SSE response"); + errno = read_result.result == 0 ? ECONNRESET : -read_result.result; state_ = client_state::disconnected; co_return false; } @@ -575,6 +622,7 @@ class sse_client { if (result == parse_result::error) { ELIO_LOG_ERROR("Failed to parse SSE response: {}", parser.error_message()); + errno = EBADMSG; state_ = client_state::disconnected; co_return false; } @@ -583,6 +631,7 @@ class sse_client { if (parser.get_status() != status::ok) { ELIO_LOG_ERROR("SSE request failed: {}", static_cast(parser.get_status())); + errno = EBADMSG; state_ = client_state::disconnected; co_return false; } @@ -601,6 +650,7 @@ class sse_client { if (parser_.failed()) { ELIO_LOG_ERROR("SSE parse error: {}", parser_.error_message()); + errno = EBADMSG; state_ = client_state::disconnected; co_return false; } @@ -611,6 +661,7 @@ class sse_client { if (response_data.size() > 8192) { ELIO_LOG_ERROR("SSE response headers too large"); + errno = EMSGSIZE; state_ = client_state::disconnected; co_return false; } diff --git a/include/elio/http/websocket_client.hpp b/include/elio/http/websocket_client.hpp index 460860a2..d7855f41 100644 --- a/include/elio/http/websocket_client.hpp +++ b/include/elio/http/websocket_client.hpp @@ -24,6 +24,7 @@ #include #include +#include #include #include #include @@ -278,7 +279,7 @@ class ws_client { } // Perform WebSocket handshake - bool success = co_await perform_handshake(); + bool success = co_await perform_handshake(std::move(token)); if (success) { state_ = connection_state::open; ELIO_LOG_DEBUG("WebSocket connected to {}{}", host_, path_); @@ -359,7 +360,7 @@ class ws_client { } /// Perform WebSocket upgrade handshake - coro::task perform_handshake() { + coro::task perform_handshake(coro::cancel_token token) { // Generate key ws_key_ = generate_websocket_key(); @@ -400,10 +401,53 @@ class ws_client { parser.set_max_headers(config_.max_headers); parser.set_max_header_size(config_.max_header_size); size_t total_read = 0; + auto* sched = runtime::scheduler::current(); + const bool deadline_enforced = + sched != nullptr && config_.read_timeout.count() > 0; + const auto response_deadline = + std::chrono::steady_clock::now() + config_.read_timeout; while (!parser.is_complete() && !parser.has_error()) { - auto read_result = co_await read(buffer_.data(), buffer_.size()); + if (token.is_cancelled()) { + errno = ECANCELED; + stream_.disconnect(); + co_return false; + } + + io::io_result read_result{}; + if (deadline_enforced) { + auto remaining = + response_deadline - std::chrono::steady_clock::now(); + if (remaining.count() <= 0) { + ELIO_LOG_ERROR("WebSocket handshake response timed out after {}s", + config_.read_timeout.count()); + errno = ETIMEDOUT; + stream_.disconnect(); + co_return false; + } + + auto timed_out = std::make_shared>(false); + coro::cancel_source watchdog_cancel; + auto watchdog = http::detail::arm_fd_shutdown_watchdog( + sched, stream_.fd(), remaining, + watchdog_cancel.get_token(), timed_out); + read_result = co_await read(buffer_.data(), buffer_.size()); + watchdog_cancel.cancel(); + co_await watchdog; + if (timed_out->load(std::memory_order_acquire)) { + stream_.mark_externally_shut_down(); + ELIO_LOG_ERROR("WebSocket handshake response timed out after {}s", + config_.read_timeout.count()); + errno = ETIMEDOUT; + stream_.disconnect(); + co_return false; + } + } else { + read_result = co_await read(buffer_.data(), buffer_.size()); + } + if (read_result.result <= 0) { ELIO_LOG_ERROR("Failed to read WebSocket handshake response"); + errno = read_result.result == 0 ? ECONNRESET : -read_result.result; co_return false; } @@ -413,18 +457,21 @@ class ws_client { if (pres == parse_result::error) { ELIO_LOG_ERROR("Failed to parse WebSocket handshake response"); + errno = EBADMSG; co_return false; } total_read += static_cast(read_result.result); if (total_read > 8192 && !parser.is_complete()) { ELIO_LOG_ERROR("WebSocket handshake response too large"); + errno = EMSGSIZE; co_return false; } } if (parser.has_error()) { ELIO_LOG_ERROR("Failed to parse WebSocket handshake response"); + errno = EBADMSG; co_return false; } @@ -432,6 +479,7 @@ class ws_client { if (parser.get_status() != status::switching_protocols) { ELIO_LOG_ERROR("WebSocket handshake failed: {}", static_cast(parser.get_status())); + errno = EBADMSG; co_return false; } @@ -439,6 +487,7 @@ class ws_client { auto accept = parser.get_headers().get("Sec-WebSocket-Accept"); if (!verify_websocket_accept(accept, ws_key_)) { ELIO_LOG_ERROR("Invalid Sec-WebSocket-Accept header"); + errno = EBADMSG; co_return false; } diff --git a/include/elio/net/stream.hpp b/include/elio/net/stream.hpp index c576fadf..81c61235 100644 --- a/include/elio/net/stream.hpp +++ b/include/elio/net/stream.hpp @@ -183,6 +183,21 @@ class stream { stream_ = std::monostate{}; } + /// Mark the active TLS stream as externally shut down. + /// + /// Timeout watchdogs may interrupt an in-flight TLS read/write by calling + /// shutdown(2) on the file descriptor from another coroutine. After the + /// I/O operation returns and the watchdog has been joined, call this before + /// destroying the stream so tls_stream skips SSL_shutdown on the unusable + /// socket. Plain TCP streams do not need any extra bookkeeping. + void mark_externally_shut_down() noexcept { +#if defined(ELIO_HAS_TLS) && ELIO_HAS_TLS + if (auto* tls = std::get_if(&stream_)) { + tls->mark_externally_shut_down(); + } +#endif + } + /// Get last use time (for connection pooling) std::chrono::steady_clock::time_point last_use() const noexcept { return last_use_; diff --git a/tests/unit/test_http_client.cpp b/tests/unit/test_http_client.cpp index fe1fd51b..57570b74 100644 --- a/tests/unit/test_http_client.cpp +++ b/tests/unit/test_http_client.cpp @@ -5,14 +5,18 @@ // 4. A connection with leftover bytes after the response is NOT pooled // (response-splitting prevention). // 5. Repeated informational responses cannot bypass max_response_size. +// 6. WebSocket and SSE client handshakes honor read_timeout while waiting +// for protocol response headers. // -// These exercise the post-PR send_request path. We stand up a tiny TCP -// listener that pretends to be an HTTP server and craft hand-built byte -// sequences for each scenario — no real http::server in the loop. +// These stand up a tiny TCP listener that pretends to be an HTTP-based server +// and craft hand-built byte sequences for each scenario -- no real +// http::server in the loop. #include #include +#include +#include #include #include #include @@ -43,6 +47,11 @@ std::string make_https_url(uint16_t port, std::string_view path = "/") { std::string(path); } +std::string make_ws_url(uint16_t port, std::string_view path = "/ws") { + return std::string("ws://127.0.0.1:") + std::to_string(port) + + std::string(path); +} + // Drain a single request from the stream (until "\r\n\r\n"). Used by the // fake server to know when to start writing its response. task drain_request_headers(elio::net::tcp_stream& s) { @@ -174,6 +183,117 @@ TEST_CASE("HTTP client read_timeout fires on a stalled server", REQUIRE(client_errno == ETIMEDOUT); } +TEST_CASE("WebSocket client read_timeout fires on a stalled handshake response", + "[websocket][client][timeout][regression]") { + auto listener = tcp_listener::bind(ipv4_address("127.0.0.1", 0)); + REQUIRE(listener.has_value()); + uint16_t port = listener->local_address().port(); + + scheduler sched(2); + sched.start(); + + std::atomic client_done{false}; + std::atomic client_failed{false}; + std::atomic client_errno{0}; + std::atomic client_elapsed_ms{-1}; + + // Server: accept and drain the WebSocket upgrade request, but never send + // the 101 response. The client's read_timeout must bound this phase. + sched.go([&]() -> task { + auto stream = co_await listener->accept(); + REQUIRE(stream.has_value()); + co_await drain_request_headers(*stream); + co_await elio::time::sleep_for(std::chrono::seconds(3)); + stream->shutdown_socket(); + }); + + sched.go([&]() -> task { + elio::http::websocket::client_config cfg; + cfg.read_timeout = std::chrono::seconds(1); + elio::http::websocket::ws_client client(cfg); + + auto t0 = std::chrono::steady_clock::now(); + bool ok = co_await client.connect(make_ws_url(port)); + auto elapsed = std::chrono::steady_clock::now() - t0; + + client_elapsed_ms = std::chrono::duration_cast( + elapsed).count(); + if (!ok) { + client_failed = true; + client_errno = errno; + } + client_done = true; + }); + + for (int i = 0; i < 500 && !client_done; ++i) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + + sched.shutdown(); + + REQUIRE(client_done); + REQUIRE(client_failed); + REQUIRE(client_errno == ETIMEDOUT); + REQUIRE(client_elapsed_ms.load() >= 0); + REQUIRE(client_elapsed_ms.load() < 3000); +} + +TEST_CASE("SSE client read_timeout fires on stalled response headers", + "[sse][client][timeout][regression]") { + auto listener = tcp_listener::bind(ipv4_address("127.0.0.1", 0)); + REQUIRE(listener.has_value()); + uint16_t port = listener->local_address().port(); + + scheduler sched(2); + sched.start(); + + std::atomic client_done{false}; + std::atomic client_failed{false}; + std::atomic client_errno{0}; + std::atomic client_elapsed_ms{-1}; + + // Server: accept and drain the SSE GET request, but never send HTTP + // headers. The client's read_timeout must bound response header reads. + sched.go([&]() -> task { + auto stream = co_await listener->accept(); + REQUIRE(stream.has_value()); + co_await drain_request_headers(*stream); + co_await elio::time::sleep_for(std::chrono::seconds(3)); + stream->shutdown_socket(); + }); + + sched.go([&]() -> task { + elio::http::sse::client_config cfg; + cfg.auto_reconnect = false; + cfg.read_timeout = std::chrono::seconds(1); + elio::http::sse::sse_client client(cfg); + + auto t0 = std::chrono::steady_clock::now(); + bool ok = co_await client.connect(make_url(port, "/events")); + auto elapsed = std::chrono::steady_clock::now() - t0; + + client_elapsed_ms = std::chrono::duration_cast( + elapsed).count(); + if (!ok) { + client_failed = true; + client_errno = errno; + } + client_done = true; + }); + + for (int i = 0; i < 500 && !client_done; ++i) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + + sched.shutdown(); + + REQUIRE(client_done); + REQUIRE(client_failed); + REQUIRE(client_errno == ETIMEDOUT); + REQUIRE(client_elapsed_ms.load() >= 0); + REQUIRE(client_elapsed_ms.load() < 3000); +} + #if defined(ELIO_HAS_TLS) && ELIO_HAS_TLS TEST_CASE("HTTP client connect_timeout fires on a stalled TLS handshake", "[http][client][timeout][tls]") { diff --git a/wiki/API-Reference.md b/wiki/API-Reference.md index 06707c2f..ffb17dee 100644 --- a/wiki/API-Reference.md +++ b/wiki/API-Reference.md @@ -1264,6 +1264,12 @@ struct client_config : base_client_config { }; ``` +`websocket::client_config` and `sse::client_config` also inherit +`base_client_config`: `connect_timeout` bounds TCP connect and TLS handshake, +while `read_timeout` bounds the WebSocket upgrade response and SSE response +header read performed by `connect()`. A value less than or equal to zero +disables these client-side read deadlines. + ### `server_config` ```cpp diff --git a/wiki/WebSocket-SSE.md b/wiki/WebSocket-SSE.md index d1a1affd..48f8ffaf 100644 --- a/wiki/WebSocket-SSE.md +++ b/wiki/WebSocket-SSE.md @@ -84,6 +84,8 @@ coro::task connect_example() { // Create client client_config config; config.subprotocols = {"chat", "json"}; // Optional subprotocols + config.connect_timeout = std::chrono::seconds(10); + config.read_timeout = std::chrono::seconds(30); // Upgrade response deadline ws_client client(config); @@ -213,6 +215,8 @@ coro::task listen_events() { client_config config; config.auto_reconnect = true; config.default_retry_ms = 3000; + config.connect_timeout = std::chrono::seconds(10); + config.read_timeout = std::chrono::seconds(30); // Response header deadline sse_client client(config); @@ -235,6 +239,12 @@ coro::task listen_events() { } ``` +For WebSocket and SSE clients, `connect_timeout` bounds TCP connect and TLS +handshake setup. `read_timeout` bounds the protocol response headers read by +`connect()` -- the WebSocket `101 Switching Protocols` response or the SSE +`text/event-stream` response headers. Values less than or equal to zero disable +these client-side read deadlines. + ### SSE Event Format SSE events are formatted as text with specific fields: From 7b96d3c7a4444b1b3d4e87f15a8ff097bb3e6661 Mon Sep 17 00:00:00 2001 From: Coldwings Date: Sat, 11 Jul 2026 10:31:06 +0800 Subject: [PATCH 2/2] Clean up failed client handshake streams --- include/elio/http/sse_client.hpp | 38 ++++++++++++-------------- include/elio/http/websocket_client.hpp | 12 ++++++++ 2 files changed, 29 insertions(+), 21 deletions(-) diff --git a/include/elio/http/sse_client.hpp b/include/elio/http/sse_client.hpp index 2b3ae5b8..4f02dd41 100644 --- a/include/elio/http/sse_client.hpp +++ b/include/elio/http/sse_client.hpp @@ -498,6 +498,14 @@ class sse_client { co_return false; } stream_ = std::move(*conn_result); + + auto fail_connect = [&]() noexcept { + int saved_errno = errno; + stream_.disconnect(); + errno = saved_errno; + state_ = client_state::disconnected; + return false; + }; // Send HTTP request std::string request; @@ -527,8 +535,7 @@ class sse_client { auto send_result = co_await write_exactly(request.data(), request.size()); if (send_result.result != static_cast(request.size())) { ELIO_LOG_ERROR("Failed to send SSE request"); - state_ = client_state::disconnected; - co_return false; + co_return fail_connect(); } // Read response headers. @@ -553,9 +560,7 @@ class sse_client { while (true) { if (token_.is_cancelled()) { errno = ECANCELED; - stream_.disconnect(); - state_ = client_state::disconnected; - co_return false; + co_return fail_connect(); } io::io_result read_result{}; @@ -566,9 +571,7 @@ class sse_client { ELIO_LOG_ERROR("SSE response headers timed out after {}s", config_.read_timeout.count()); errno = ETIMEDOUT; - stream_.disconnect(); - state_ = client_state::disconnected; - co_return false; + co_return fail_connect(); } auto timed_out = std::make_shared>(false); @@ -584,9 +587,7 @@ class sse_client { ELIO_LOG_ERROR("SSE response headers timed out after {}s", config_.read_timeout.count()); errno = ETIMEDOUT; - stream_.disconnect(); - state_ = client_state::disconnected; - co_return false; + co_return fail_connect(); } } else { read_result = co_await read(buffer_.data(), buffer_.size()); @@ -595,8 +596,7 @@ class sse_client { if (read_result.result <= 0) { ELIO_LOG_ERROR("Failed to read SSE response"); errno = read_result.result == 0 ? ECONNRESET : -read_result.result; - state_ = client_state::disconnected; - co_return false; + co_return fail_connect(); } response_data.append(buffer_.data(), static_cast(read_result.result)); @@ -623,8 +623,7 @@ class sse_client { ELIO_LOG_ERROR("Failed to parse SSE response: {}", parser.error_message()); errno = EBADMSG; - state_ = client_state::disconnected; - co_return false; + co_return fail_connect(); } // Check status code @@ -632,8 +631,7 @@ class sse_client { ELIO_LOG_ERROR("SSE request failed: {}", static_cast(parser.get_status())); errno = EBADMSG; - state_ = client_state::disconnected; - co_return false; + co_return fail_connect(); } // Check content type @@ -651,8 +649,7 @@ class sse_client { ELIO_LOG_ERROR("SSE parse error: {}", parser_.error_message()); errno = EBADMSG; - state_ = client_state::disconnected; - co_return false; + co_return fail_connect(); } } @@ -662,8 +659,7 @@ class sse_client { if (response_data.size() > 8192) { ELIO_LOG_ERROR("SSE response headers too large"); errno = EMSGSIZE; - state_ = client_state::disconnected; - co_return false; + co_return fail_connect(); } } diff --git a/include/elio/http/websocket_client.hpp b/include/elio/http/websocket_client.hpp index d7855f41..d8fee5d4 100644 --- a/include/elio/http/websocket_client.hpp +++ b/include/elio/http/websocket_client.hpp @@ -229,8 +229,11 @@ class ws_client { /// Internal connect implementation coro::task connect_impl(std::string_view url_str, coro::cancel_token token) { + state_ = connection_state::connecting; + // Check if already cancelled if (token.is_cancelled()) { + state_ = connection_state::closed; co_return false; } @@ -238,6 +241,7 @@ class ws_client { auto parsed = parse_ws_url(url_str); if (!parsed) { ELIO_LOG_ERROR("Invalid WebSocket URL: {}", url_str); + state_ = connection_state::closed; co_return false; } @@ -255,6 +259,7 @@ class ws_client { // Check cancellation before connection if (token.is_cancelled()) { + state_ = connection_state::closed; co_return false; } @@ -268,6 +273,7 @@ class ws_client { config_.rotate_resolved_addresses, config_.connect_timeout); if (!conn_result) { + state_ = connection_state::closed; co_return false; } stream_ = std::move(*conn_result); @@ -275,6 +281,7 @@ class ws_client { // Check cancellation before handshake if (token.is_cancelled()) { stream_.disconnect(); + state_ = connection_state::closed; co_return false; } @@ -283,6 +290,11 @@ class ws_client { if (success) { state_ = connection_state::open; ELIO_LOG_DEBUG("WebSocket connected to {}{}", host_, path_); + } else { + int saved_errno = errno; + stream_.disconnect(); + errno = saved_errno; + state_ = connection_state::closed; } co_return success;