Skip to content

Bound WebSocket and SSE client handshakes by read_timeout #482

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Coldwings merged 2 commits into from
Jul 11, 2026
Merged

Bound WebSocket and SSE client handshakes by read_timeout #482

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
0b879bb
Bound client protocol handshakes by read_timeout
Coldwings Jul 11, 2026
7b96d3c
Clean up failed client handshake streams
Coldwings Jul 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 28 additions & 0 deletions include/elio/http/client_base.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,13 +18,16 @@
#include <elio/runtime/scheduler.hpp>
#include <elio/time/timer.hpp>

#include <sys/socket.h>

#include <atomic>
#include <string>
#include <chrono>
#include <memory>
#include <mutex>
#include <optional>
#include <unordered_map>
#include <utility>

namespace elio::http {

Expand All @@ -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<void>
arm_fd_shutdown_watchdog(runtime::scheduler* sched,
int fd,
std::chrono::nanoseconds timeout,
coro::cancel_token watchdog_token,
std::shared_ptr<std::atomic<bool>> timed_out) {
return sched->go_joinable(
[fd, timeout, tok = std::move(watchdog_token),
flag = std::move(timed_out)]() -> coro::task<void> {
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
Expand Down
73 changes: 60 additions & 13 deletions include/elio/http/sse_client.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@

#include <string>
#include <string_view>
#include <cerrno>
#include <memory>
#include <optional>
#include <vector>
Expand Down Expand Up @@ -497,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;
Expand Down Expand Up @@ -526,8 +535,7 @@ class sse_client {
auto send_result = co_await write_exactly(request.data(), request.size());
if (send_result.result != static_cast<ssize_t>(request.size())) {
ELIO_LOG_ERROR("Failed to send SSE request");
state_ = client_state::disconnected;
co_return false;
co_return fail_connect();
}

// Read response headers.
Expand All @@ -543,13 +551,52 @@ 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;
co_return fail_connect();
}

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;
co_return fail_connect();
}

auto timed_out = std::make_shared<std::atomic<bool>>(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;
co_return fail_connect();
}
} else {
read_result = co_await read(buffer_.data(), buffer_.size());
}

if (read_result.result <= 0) {
ELIO_LOG_ERROR("Failed to read SSE response");
state_ = client_state::disconnected;
co_return false;
errno = read_result.result == 0 ? ECONNRESET : -read_result.result;
co_return fail_connect();
}

response_data.append(buffer_.data(), static_cast<size_t>(read_result.result));
Expand All @@ -575,16 +622,16 @@ class sse_client {
if (result == parse_result::error) {
ELIO_LOG_ERROR("Failed to parse SSE response: {}",
parser.error_message());
state_ = client_state::disconnected;
co_return false;
errno = EBADMSG;
co_return fail_connect();
}

// Check status code
if (parser.get_status() != status::ok) {
ELIO_LOG_ERROR("SSE request failed: {}",
static_cast<int>(parser.get_status()));
state_ = client_state::disconnected;
co_return false;
errno = EBADMSG;
co_return fail_connect();
}

// Check content type
Expand All @@ -601,8 +648,8 @@ class sse_client {
if (parser_.failed()) {
ELIO_LOG_ERROR("SSE parse error: {}",
parser_.error_message());
state_ = client_state::disconnected;
co_return false;
errno = EBADMSG;
co_return fail_connect();
}
}

Expand All @@ -611,8 +658,8 @@ class sse_client {

if (response_data.size() > 8192) {
ELIO_LOG_ERROR("SSE response headers too large");
state_ = client_state::disconnected;
co_return false;
errno = EMSGSIZE;
co_return fail_connect();
}
}

Expand Down
67 changes: 64 additions & 3 deletions include/elio/http/websocket_client.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@

#include <string>
#include <string_view>
#include <cerrno>
#include <memory>
#include <optional>
#include <vector>
Expand Down Expand Up @@ -228,15 +229,19 @@ class ws_client {

/// Internal connect implementation
coro::task<bool> 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;
}

// Parse URL
auto parsed = parse_ws_url(url_str);
if (!parsed) {
ELIO_LOG_ERROR("Invalid WebSocket URL: {}", url_str);
state_ = connection_state::closed;
co_return false;
}

Expand All @@ -254,6 +259,7 @@ class ws_client {

// Check cancellation before connection
if (token.is_cancelled()) {
state_ = connection_state::closed;
co_return false;
}

Expand All @@ -267,21 +273,28 @@ 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);

// Check cancellation before handshake
if (token.is_cancelled()) {
stream_.disconnect();
state_ = connection_state::closed;
co_return false;
}

// 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_);
Comment thread
Coldwings marked this conversation as resolved.
} else {
int saved_errno = errno;
stream_.disconnect();
errno = saved_errno;
state_ = connection_state::closed;
}

co_return success;
Expand Down Expand Up @@ -359,7 +372,7 @@ class ws_client {
}

/// Perform WebSocket upgrade handshake
coro::task<bool> perform_handshake() {
coro::task<bool> perform_handshake(coro::cancel_token token) {
// Generate key
ws_key_ = generate_websocket_key();

Expand Down Expand Up @@ -400,10 +413,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<std::atomic<bool>>(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;
}

Expand All @@ -413,32 +469,37 @@ 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<size_t>(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;
}

// Check status code
if (parser.get_status() != status::switching_protocols) {
ELIO_LOG_ERROR("WebSocket handshake failed: {}",
static_cast<int>(parser.get_status()));
errno = EBADMSG;
co_return false;
}

// Verify Sec-WebSocket-Accept
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;
}

Expand Down
15 changes: 15 additions & 0 deletions include/elio/net/stream.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<tls::tls_stream>(&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_;
Expand Down
Loading