From 297749f2c3ed42dfdecc7658b2199643d83b3c7a Mon Sep 17 00:00:00 2001 From: uartnet <140632163+uartnet@users.noreply.github.com> Date: Sat, 15 Aug 2026 01:17:16 +0200 Subject: [PATCH 1/9] fix(io): harden asynchronous stream lifecycles --- TODO.md | 15 ++ bin/ncat/bin/main.cpp | 20 +- bin/ncat/lib/rstream/ncat/client.cpp | 15 +- bin/ncat/lib/rstream/ncat/server.cpp | 13 +- bin/tunnel/lib/rstream/tunnel/proxy.cpp | 81 ++++++- lib/rstream/core/completion_handler.hpp | 8 + lib/rstream/core/windows/blocking_handle.cpp | 6 +- lib/rstream/io-rstrm/acceptor.cpp | 5 +- lib/rstream/io-rstrm/socket.cpp | 60 +++-- lib/rstream/io-rstrm/socket.hpp | 2 + lib/rstream/io/acceptor_base.hpp | 3 +- .../io/detail/stream/acceptor_impl.hpp | 4 +- lib/rstream/io/detail/stream/acceptor_ssl.cpp | 9 +- .../io/detail/stream/stream_socket.cpp | 16 ++ .../io/detail/stream/stream_socket.hpp | 2 + .../io/detail/stream/stream_socket_impl.hpp | 26 ++ .../io/detail/stream/stream_socket_ssl.cpp | 115 ++++++++- .../io/detail/stream/stream_socket_ssl.hpp | 2 + lib/rstream/io/queue.hpp | 12 +- lib/rstream/io/stream_socket_base.hpp | 24 ++ .../core/common/test_core_executor_binder.cpp | 41 ++++ test/io/common/test_io_common_queue.cpp | 67 ++++++ test/io/common/test_io_common_stream_tls.cpp | 227 ++++++++++++++++++ test/io/io-rstrm/test_io_rstrm_handshake.cpp | 12 + test/ncat/CMakeLists.txt | 5 + test/ncat/test_ncat_cli_runtime.py | 66 +++++ test/tunnel/test_tunnel_proxy.cpp | 29 ++- 27 files changed, 809 insertions(+), 76 deletions(-) create mode 100644 TODO.md create mode 100644 test/ncat/test_ncat_cli_runtime.py diff --git a/TODO.md b/TODO.md new file mode 100644 index 0000000..6cc093a --- /dev/null +++ b/TODO.md @@ -0,0 +1,15 @@ +# Follow-up work + +## C++ TLS and asynchronous lifecycle hardening + +The focused multithreaded TLS half-close test passes with the quality, AddressSanitizer, and ThreadSanitizer builds as of 2026-08-13. The demonstrated memory-safety defect was an allocator-lifetime issue at type-erased asynchronous-operation boundaries, not an OpenSSL or TLS defect. The affected boundaries have dedicated fixes and regression tests in the current worktree. + +This area is deliberately not the critical path of the current guide and sample validation. Before declaring the TLS lifecycle audit exhaustive, complete the following follow-up matrix: + +- Exercise TLS 1.2 and TLS 1.3 handshake, cancellation, half-close, peer-close, timeout, and abrupt-reset paths with one and several `io_context` worker threads. +- Repeat client/server interoperability in both directions for Go and C++, including concurrent streams and shutdown during backpressure. +- Run long-duration and high-concurrency stress tests under AddressSanitizer and ThreadSanitizer, and retain machine-readable evidence in CI. +- Add the corresponding stateful associated-allocator abandonment checks to every remaining type-erased operation boundary, including Windows-only handles in Windows CI. +- Remove the stale OpenSSL 1.1 linker search path emitted by macOS builds after verifying that packaging remains compatible with the supported OpenSSL versions. + +Completion requires zero sanitizer findings, deterministic cancellation, exactly-once completion, no handler after owner destruction, bounded shutdown time, and no measurable throughput or latency regression against the recorded baseline. diff --git a/bin/ncat/bin/main.cpp b/bin/ncat/bin/main.cpp index d18a1f6..b5926da 100644 --- a/bin/ncat/bin/main.cpp +++ b/bin/ncat/bin/main.cpp @@ -3,6 +3,7 @@ #include #include #include +#include #include #include @@ -48,9 +49,26 @@ this program is distributed with the rstream C++ tools. See https://rstream.io/d const auto version = std::string("rstream-ncat ") + RSTREAM_VERSION; +static std::vector normalize_cli_arguments(int argc, char** argv) +{ + std::vector result; + for (int index = 1; index < argc; ++index) { + std::string argument = argv[index]; + if (argument.starts_with("-c=") || argument.starts_with("-e=")) { + result.emplace_back(argument.substr(0, 2)); + result.emplace_back(argument.substr(3)); + } + else { + result.emplace_back(std::move(argument)); + } + } + return result; +} + int run(int argc, char** argv) { - auto args = docopt::docopt(USAGE, {argv + 1, argv + argc}, true, version); + auto cli_args = normalize_cli_arguments(argc, argv); + auto args = docopt::docopt(USAGE, cli_args, true, version); bool verbose = false; { auto it = args.find("--verbose"); diff --git a/bin/ncat/lib/rstream/ncat/client.cpp b/bin/ncat/lib/rstream/ncat/client.cpp index 9e76fc5..ff725f9 100644 --- a/bin/ncat/lib/rstream/ncat/client.cpp +++ b/bin/ncat/lib/rstream/ncat/client.cpp @@ -455,9 +455,18 @@ void client::impl::on_read_std_in(const boost::system::error_code& error_code, s } else if (eos) { m_std_in_eos = true; -#ifndef RSTREAM_WITH_IO_STREAMS - boost::system::error_code tmp; - m_socket.shutdown(boost::asio::socket_base::shutdown_send, tmp); +#ifdef RSTREAM_WITH_IO_STREAMS + m_socket.async_shutdown_send(boost::asio::bind_executor(m_strand, [self = shared_from_this()](const boost::system::error_code& shutdown_error) { + if (self->m_state == state::connected && shutdown_error && !core::helpers::is_eof_error(shutdown_error)) { + self->on_error(shutdown_error); + } + })); +#else + boost::system::error_code shutdown_error; + m_socket.shutdown(boost::asio::socket_base::shutdown_send, shutdown_error); + if (shutdown_error) { + on_error(shutdown_error); + } #endif } else { diff --git a/bin/ncat/lib/rstream/ncat/server.cpp b/bin/ncat/lib/rstream/ncat/server.cpp index 478bcf7..067d6c3 100644 --- a/bin/ncat/lib/rstream/ncat/server.cpp +++ b/bin/ncat/lib/rstream/ncat/server.cpp @@ -389,7 +389,7 @@ class RSTREAM_GNUC_INTERNAL server::impl::session_proxy : public session, public class RSTREAM_GNUC_INTERNAL server::impl::session_exec : public session, public std::enable_shared_from_this { public: - session_exec(socket_type&& downstream_socket, const settings_server& settings, const session_id_type& session_id, const exec& exec, bool downstream_half_close); + session_exec(socket_type&& downstream_socket, const settings_server& settings, const session_id_type& session_id, const exec& exec); void async_run(async_run_completion_handler&& handler) override; @@ -454,8 +454,6 @@ class RSTREAM_GNUC_INTERNAL server::impl::session_exec : public session, public const exec m_exec; - const bool m_downstream_half_close; - core::logger m_logger; state m_state; @@ -726,7 +724,7 @@ void server::impl::on_accept(const boost::system::error_code& error_code) session_ptr = std::make_shared(std::move(m_socket), m_settings, session_id, boost::get(m_config.m_remote)); } else if (m_config.m_remote.type() == typeid(exec)) { - session_ptr = std::make_shared(std::move(m_socket), m_settings, session_id, boost::get(m_config.m_remote), m_config.m_local.m_url.scheme() == "tcp"); + session_ptr = std::make_shared(std::move(m_socket), m_settings, session_id, boost::get(m_config.m_remote)); } if (session_ptr) { m_sessions.insert(std::make_pair(session_id, session_ptr)); @@ -1172,14 +1170,13 @@ void server::impl::session_proxy::on_close(const boost::system::error_code& erro m_buffer_read_upstream = nullptr; } -server::impl::session_exec::session_exec(socket_type&& downstream_socket, const settings_server& settings, const session_id_type& session_id, const exec& exec, bool downstream_half_close) +server::impl::session_exec::session_exec(socket_type&& downstream_socket, const settings_server& settings, const session_id_type& session_id, const exec& exec) : m_executor(downstream_socket.get_executor()), m_strand(m_executor), m_settings(settings), m_downstream_socket(std::move(downstream_socket)), m_session_id(session_id), m_exec(exec), - m_downstream_half_close(downstream_half_close), m_logger({"rstream", "ncat", "session", fmt::format("#{}", session_id)}), m_state(state::null), m_child_stdin(get_io_context(m_executor)), @@ -1409,10 +1406,6 @@ void server::impl::session_exec::on_read_downstream(const boost::system::error_c } if (error_code) { if (core::helpers::is_eof_error(error_code)) { - if (!m_downstream_half_close) { - on_close(boost::system::error_code()); - return; - } { boost::system::error_code tmp; m_child_stdin.close(tmp); diff --git a/bin/tunnel/lib/rstream/tunnel/proxy.cpp b/bin/tunnel/lib/rstream/tunnel/proxy.cpp index bc0400e..9671662 100644 --- a/bin/tunnel/lib/rstream/tunnel/proxy.cpp +++ b/bin/tunnel/lib/rstream/tunnel/proxy.cpp @@ -189,6 +189,10 @@ class RSTREAM_GNUC_INTERNAL proxy::impl::session : public std::enable_shared_fro void on_read(const boost::system::error_code& error_code, std::size_t size, type type); + void do_shutdown_send(type type); + + void on_shutdown_send(const boost::system::error_code& error_code); + void do_write(type type); void on_write(const boost::system::error_code& error_code, std::size_t size, type type); @@ -221,6 +225,10 @@ class RSTREAM_GNUC_INTERNAL proxy::impl::session : public std::enable_shared_fro state m_state; + bool m_downstream_read_closed; + + bool m_upstream_read_closed; + async_run_completion_handler m_handler; std::shared_ptr m_buffer_read_downstream; @@ -574,7 +582,9 @@ proxy::impl::session::session(downstream_socket_type&& downstream_socket, const m_session_id(session_id), m_upstream_address(upstream_address), m_logger({"rstream", "tunnel", "session", fmt::format("#{}", session_id)}), - m_state(state::null) + m_state(state::null), + m_downstream_read_closed(false), + m_upstream_read_closed(false) { std::stringstream str; { @@ -790,14 +800,71 @@ void proxy::impl::session::on_read(const boost::system::error_code& error_code, if (m_state != state::connected) { return; } - if (error_code) { + const auto eof = error_code && core::helpers::is_eof_error(error_code); + if (error_code && !eof) { on_error(error_code); + return; } - else { + if (eof) { + if (type == type::downstream) { + m_downstream_read_closed = true; + } + else { + m_upstream_read_closed = true; + } + } + if (size > 0) { auto& buffer = type == type::downstream ? *m_buffer_read_downstream : *m_buffer_read_upstream; buffer.set_size(size); do_write(type == type::downstream ? type::upstream : type::downstream); } + else if (eof) { + do_shutdown_send(type == type::downstream ? type::upstream : type::downstream); + } + else { + do_read(type); + } +} + +void proxy::impl::session::do_shutdown_send(type type) +{ +#ifdef DEBUG_BUILD + assert(m_strand.running_in_this_thread()); +#endif + if (type == type::downstream) { + auto completion_handler = boost::asio::bind_executor( + m_strand, + [ptr = shared_from_this()](const boost::system::error_code& error_code) { ptr->on_shutdown_send(error_code); }); + m_downstream_socket.async_shutdown_send(std::move(completion_handler)); + } + else { +#ifdef RSTREAM_WITH_IO_STREAMS + auto completion_handler = boost::asio::bind_executor( + m_strand, + [ptr = shared_from_this()](const boost::system::error_code& error_code) { ptr->on_shutdown_send(error_code); }); + m_upstream_socket.async_shutdown_send(std::move(completion_handler)); +#else + boost::system::error_code error_code; + m_upstream_socket.shutdown(boost::asio::socket_base::shutdown_send, error_code); + on_shutdown_send(error_code); +#endif + } +} + +void proxy::impl::session::on_shutdown_send(const boost::system::error_code& error_code) +{ +#ifdef DEBUG_BUILD + assert(m_strand.running_in_this_thread()); +#endif + if (m_state != state::connected) { + return; + } + if (error_code && !core::helpers::is_eof_error(error_code)) { + on_error(error_code); + } + else if (m_downstream_read_closed && m_upstream_read_closed) { + on_close(boost::system::error_code()); + } } void proxy::impl::session::do_write(type type) @@ -831,7 +898,13 @@ void proxy::impl::session::on_write(const boost::system::error_code& error_code, on_error(error_code); } else { - do_read(type == type::downstream ? type::upstream : type::downstream); + const auto source_closed = type == type::downstream ? m_upstream_read_closed : m_downstream_read_closed; + if (source_closed) { + do_shutdown_send(type); + } + else { + do_read(type == type::downstream ? type::upstream : type::downstream); + } } } diff --git a/lib/rstream/core/completion_handler.hpp b/lib/rstream/core/completion_handler.hpp index df6d386..74435da 100644 --- a/lib/rstream/core/completion_handler.hpp +++ b/lib/rstream/core/completion_handler.hpp @@ -152,6 +152,14 @@ void invoke_completion_handler(const Executor& ctx, Handler&& handler, Args&&... boost::asio::post(ctx, invocation_type(ctx, std::forward(handler), std::forward(args)...)); } +template +void dispatch_completion_handler(const Executor& ctx, Handler&& handler, Args&&... args) +{ + using invocation_type = detail::completion_invocation, std::decay_t, std::decay_t...>; + const auto executor = boost::asio::get_associated_executor(handler, ctx); + boost::asio::dispatch(executor, invocation_type(ctx, std::forward(handler), std::forward(args)...)); +} + } // namespace core } // namespace rstream diff --git a/lib/rstream/core/windows/blocking_handle.cpp b/lib/rstream/core/windows/blocking_handle.cpp index ba4771c..24e0513 100644 --- a/lib/rstream/core/windows/blocking_handle.cpp +++ b/lib/rstream/core/windows/blocking_handle.cpp @@ -95,14 +95,12 @@ class blocking_handle::impl : public std::enable_shared_from_this { void async_read_some(const boost::asio::mutable_buffer& buffer, completion_handler&& handler) { - auto operation_allocator = boost::asio::get_associated_allocator(handler); - submit(std::allocate_shared(operation_allocator, operation::type::read, buffer, m_executor, std::move(handler))); + submit(std::make_shared(operation::type::read, buffer, m_executor, std::move(handler))); } void async_write(const boost::asio::const_buffer& buffer, completion_handler&& handler) { - auto operation_allocator = boost::asio::get_associated_allocator(handler); - submit(std::allocate_shared(operation_allocator, operation::type::write, buffer, m_executor, std::move(handler))); + submit(std::make_shared(operation::type::write, buffer, m_executor, std::move(handler))); } void cancel() diff --git a/lib/rstream/io-rstrm/acceptor.cpp b/lib/rstream/io-rstrm/acceptor.cpp index 2af38bd..f615a72 100644 --- a/lib/rstream/io-rstrm/acceptor.cpp +++ b/lib/rstream/io-rstrm/acceptor.cpp @@ -341,8 +341,7 @@ void acceptor::impl::async_accept(socket& peer, endpoint& endpoint, async_accept if (!handler) { return; } - auto allocator = boost::asio::get_associated_allocator(handler); - const auto op = std::allocate_shared(allocator, peer, endpoint, std::move(handler)); + const auto op = std::allocate_shared(core::allocator::wrapper(m_allocator), peer, endpoint, std::move(handler)); { std::lock_guard lock(m_mutex); if (m_start_pending) { @@ -387,7 +386,7 @@ void acceptor::impl::async_accept(socket& peer, endpoint& endpoint, async_accept boost::asio::dispatch( m_strand, core::bind_handler_allocator( - allocator, + core::allocator::wrapper(m_allocator), [self = shared_from_this(), op] { self->async_accept_internal(op); })); } diff --git a/lib/rstream/io-rstrm/socket.cpp b/lib/rstream/io-rstrm/socket.cpp index df831ba..6da181a 100644 --- a/lib/rstream/io-rstrm/socket.cpp +++ b/lib/rstream/io-rstrm/socket.cpp @@ -58,6 +58,8 @@ class RSTREAM_GNUC_INTERNAL socket::impl : public std::enable_shared_from_thisasync_read_some(buffer, std::move(handler)); } +void socket::async_shutdown_send_internal(async_shutdown_send_completion_handler&& handler) +{ + return ptr()->async_shutdown_send(std::move(handler)); +} + void socket::adopt_impl(socket&& other) noexcept { if (this != &other) { @@ -421,8 +428,7 @@ void socket::impl::async_connect(type type, const endpoint& endpoint, async_conn if (!handler) { return; } - auto allocator = boost::asio::get_associated_allocator(handler); - const auto op = std::allocate_shared(allocator, std::move(handler)); + const auto op = std::allocate_shared(core::allocator::wrapper(m_allocator), std::move(handler)); { std::lock_guard lock(m_mutex); if (m_is_state_non_null) { @@ -461,7 +467,7 @@ void socket::impl::async_connect(type type, const endpoint& endpoint, async_conn boost::asio::dispatch( m_strand, core::bind_handler_allocator( - allocator, + core::allocator::wrapper(m_allocator), [self = shared_from_this(), type, endpoint, op] { self->async_connect_internal(type, endpoint, op); })); } @@ -470,13 +476,12 @@ void socket::impl::async_write_some(const boost::asio::const_buffer& buffer, asy if (!handler) { return; } - auto allocator = boost::asio::get_associated_allocator(handler); - const auto op = std::allocate_shared(allocator, std::move(handler)); + const auto op = std::allocate_shared(core::allocator::wrapper(m_allocator), std::move(handler)); install_transfer_cancellation(op); boost::asio::dispatch( m_strand, core::bind_handler_allocator( - allocator, + core::allocator::wrapper(m_allocator), [self = shared_from_this(), buffer, op] { self->async_write_some_internal(buffer, op); })); } @@ -485,13 +490,12 @@ void socket::impl::async_write_some(const const_buffer_sequence_type& buffer, as if (!handler) { return; } - auto allocator = boost::asio::get_associated_allocator(handler); - const auto op = std::allocate_shared(allocator, std::move(handler)); + const auto op = std::allocate_shared(core::allocator::wrapper(m_allocator), std::move(handler)); install_transfer_cancellation(op); boost::asio::dispatch( m_strand, core::bind_handler_allocator( - allocator, + core::allocator::wrapper(m_allocator), [self = shared_from_this(), buffer, op] { self->async_write_some_internal(buffer, op); })); } @@ -500,13 +504,12 @@ void socket::impl::async_read_some(const boost::asio::mutable_buffer& buffer, as if (!handler) { return; } - auto allocator = boost::asio::get_associated_allocator(handler); - const auto op = std::allocate_shared(allocator, std::move(handler)); + const auto op = std::allocate_shared(core::allocator::wrapper(m_allocator), std::move(handler)); install_transfer_cancellation(op); boost::asio::dispatch( m_strand, core::bind_handler_allocator( - allocator, + core::allocator::wrapper(m_allocator), [self = shared_from_this(), buffer, op] { self->async_read_some_internal(buffer, op); })); } @@ -515,16 +518,39 @@ void socket::impl::async_read_some(const mutable_buffer_sequence_type& buffer, a if (!handler) { return; } - auto allocator = boost::asio::get_associated_allocator(handler); - const auto op = std::allocate_shared(allocator, std::move(handler)); + const auto op = std::allocate_shared(core::allocator::wrapper(m_allocator), std::move(handler)); install_transfer_cancellation(op); boost::asio::dispatch( m_strand, core::bind_handler_allocator( - allocator, + core::allocator::wrapper(m_allocator), [self = shared_from_this(), buffer, op] { self->async_read_some_internal(buffer, op); })); } +void socket::impl::async_shutdown_send(async_shutdown_send_completion_handler&& handler) +{ + if (!handler) { + return; + } + boost::asio::dispatch( + m_strand, + core::bind_handler_allocator( + core::allocator::wrapper(m_allocator), + [self = shared_from_this(), handler = std::move(handler)]() mutable { + if (self->m_state != state::connected) { + rstream::core::invoke_completion_handler(self->m_executor, std::move(handler), error::make_error_code(error::code::invalid_state)); + return; + } +#ifdef RSTREAM_WITH_IO_STREAMS + self->m_socket.async_shutdown_send(rstream::core::bind_handler_lifetime(self, std::move(handler))); +#else + boost::system::error_code error_code; + self->m_socket.shutdown(boost::asio::socket_base::shutdown_send, error_code); + rstream::core::invoke_completion_handler(self->m_executor, std::move(handler), error_code); +#endif + })); +} + void socket::impl::install_transfer_cancellation(const transfer_op::ptr& op) { auto cancellation_slot = boost::asio::get_associated_cancellation_slot(op->m_handler); @@ -826,9 +852,7 @@ void socket::impl::do_connect(const resolver_type::results_type& results) }; auto internal_handler = boost::asio::bind_executor( m_strand, - core::bind_handler_allocator( - core::allocator::wrapper(m_allocator), - boost::asio::bind_cancellation_slot(boost::asio::cancellation_slot(), std::move(completion_handler)))); + boost::asio::bind_allocator(core::allocator::wrapper(m_allocator), std::move(completion_handler))); boost::asio::async_connect(m_socket, results, std::move(internal_handler)); } diff --git a/lib/rstream/io-rstrm/socket.hpp b/lib/rstream/io-rstrm/socket.hpp index c2c3847..78cd660 100644 --- a/lib/rstream/io-rstrm/socket.hpp +++ b/lib/rstream/io-rstrm/socket.hpp @@ -74,6 +74,8 @@ class socket : public io::stream_socket_base { void async_read_some_internal(const mutable_buffer_sequence_type& buffer, async_read_some_completion_handler&& handler) override; + void async_shutdown_send_internal(async_shutdown_send_completion_handler&& handler) override; + void adopt_impl(socket&& other) noexcept; std::shared_ptr ptr(); diff --git a/lib/rstream/io/acceptor_base.hpp b/lib/rstream/io/acceptor_base.hpp index 2f5c79a..63e9fc6 100644 --- a/lib/rstream/io/acceptor_base.hpp +++ b/lib/rstream/io/acceptor_base.hpp @@ -80,8 +80,7 @@ class acceptor_base : public socket_base { [this](auto&& handler) { using operation_type = owning_accept_operation>; auto executor = io_object::get_executor(); - auto allocator = boost::asio::get_associated_allocator(handler); - auto operation = std::allocate_shared(allocator, executor, std::forward(handler)); + auto operation = std::make_shared(executor, std::forward(handler)); auto cancellation_slot = boost::asio::get_associated_cancellation_slot(operation->m_handler); if (cancellation_slot.is_connected()) { const std::weak_ptr weak_operation = operation; diff --git a/lib/rstream/io/detail/stream/acceptor_impl.hpp b/lib/rstream/io/detail/stream/acceptor_impl.hpp index 1c2d750..83f433f 100644 --- a/lib/rstream/io/detail/stream/acceptor_impl.hpp +++ b/lib/rstream/io/detail/stream/acceptor_impl.hpp @@ -211,9 +211,7 @@ native_endpoint_type acceptor_impl void acceptor_impl::async_accept_internal(stream_socket& peer, endpoint& endpoint, async_accept_completion_handler&& handler) { - const auto allocator = boost::asio::get_associated_allocator(handler); - std::allocate_shared( - allocator, + std::make_shared( std::enable_shared_from_this::shared_from_this(), peer, endpoint, m_url, std::move(handler)) ->run(); diff --git a/lib/rstream/io/detail/stream/acceptor_ssl.cpp b/lib/rstream/io/detail/stream/acceptor_ssl.cpp index 7f316ff..9e0bbd5 100644 --- a/lib/rstream/io/detail/stream/acceptor_ssl.cpp +++ b/lib/rstream/io/detail/stream/acceptor_ssl.cpp @@ -211,12 +211,11 @@ endpoint acceptor_ssl::impl::local_endpoint(boost::system::error_code& error_cod void acceptor_ssl::impl::async_accept(stream_socket& peer, endpoint& endpoint, async_accept_completion_handler&& handler) { - const auto allocator = boost::asio::get_associated_allocator(handler); - auto self = shared_from_this(); + auto self = shared_from_this(); boost::asio::dispatch( m_strand, boost::asio::bind_allocator( - allocator, + core::allocator::wrapper(m_allocator), [self = std::move(self), peer = &peer, endpoint = &endpoint, handler = std::move(handler)]() mutable { self->async_accept_internal(*peer, *endpoint, std::move(handler)); })); @@ -261,8 +260,8 @@ void acceptor_ssl::impl::async_accept_internal(stream_socket& peer, endpoint& en it = m_async_accept_upstream_ops.erase(it); } else { - const auto allocator = boost::asio::get_associated_allocator(handler); - m_async_accept_downstream_op = std::allocate_shared(allocator, peer, endpoint, std::move(handler)); + m_async_accept_downstream_op = std::allocate_shared( + core::allocator::wrapper(m_allocator), peer, endpoint, std::move(handler)); do_accept(); } } diff --git a/lib/rstream/io/detail/stream/stream_socket.cpp b/lib/rstream/io/detail/stream/stream_socket.cpp index 37e311c..c5438bf 100644 --- a/lib/rstream/io/detail/stream/stream_socket.cpp +++ b/lib/rstream/io/detail/stream/stream_socket.cpp @@ -44,6 +44,8 @@ class RSTREAM_GNUC_INTERNAL stream_socket::impl { void async_read_some(const mutable_buffer_sequence_type& buffer, async_read_some_completion_handler&& handler); + void async_shutdown_send(async_shutdown_send_completion_handler&& handler); + stream_socket_const_ptr native_handle() const; stream_socket_ptr native_handle(); @@ -149,6 +151,11 @@ void stream_socket::async_read_some_internal(const mutable_buffer_sequence_type& m_impl->async_read_some(buffer, std::move(handler)); } +void stream_socket::async_shutdown_send_internal(async_shutdown_send_completion_handler&& handler) +{ + m_impl->async_shutdown_send(std::move(handler)); +} + stream_socket::impl::impl(const executor_type& executor) : m_executor(executor) { @@ -270,6 +277,15 @@ void stream_socket::impl::async_read_some(const mutable_buffer_sequence_type& bu } } +void stream_socket::impl::async_shutdown_send(async_shutdown_send_completion_handler&& handler) +{ + if (!initialized()) { + rstream::core::invoke_completion_handler(m_executor, std::move(handler), error::make_error_code(error::code::uninitialized_object)); + return; + } + m_native_handle->async_shutdown_send(std::move(handler)); +} + stream_socket_const_ptr stream_socket::impl::native_handle() const { return m_native_handle; diff --git a/lib/rstream/io/detail/stream/stream_socket.hpp b/lib/rstream/io/detail/stream/stream_socket.hpp index 4caa320..8a74329 100644 --- a/lib/rstream/io/detail/stream/stream_socket.hpp +++ b/lib/rstream/io/detail/stream/stream_socket.hpp @@ -69,6 +69,8 @@ class stream_socket : public stream_socket_base { void async_read_some_internal(const mutable_buffer_sequence_type& buffer, async_read_some_completion_handler&& handler) override; + void async_shutdown_send_internal(async_shutdown_send_completion_handler&& handler) override; + std::shared_ptr m_impl; }; diff --git a/lib/rstream/io/detail/stream/stream_socket_impl.hpp b/lib/rstream/io/detail/stream/stream_socket_impl.hpp index d76bf53..d503588 100644 --- a/lib/rstream/io/detail/stream/stream_socket_impl.hpp +++ b/lib/rstream/io/detail/stream/stream_socket_impl.hpp @@ -3,6 +3,7 @@ #pragma once #include +#include #include #include @@ -68,6 +69,8 @@ class stream_socket_impl : public stream_socket_base, public object_ba void async_read_some_internal(const mutable_buffer_sequence_type& buffer, async_read_some_completion_handler&& handler) override; + void async_shutdown_send_internal(async_shutdown_send_completion_handler&& handler) override; + native_socket_type m_socket; const endpoint_base::protocol_type::value_type m_protocol; @@ -241,6 +244,29 @@ void stream_socket_impl::async_read_so std::move(handler))); } +template +void stream_socket_impl::async_shutdown_send_internal(async_shutdown_send_completion_handler&& handler) +{ + if constexpr (requires(native_socket_type& socket, async_shutdown_send_completion_handler&& completion_handler) { + socket.async_shutdown_send(std::move(completion_handler)); + }) { + m_socket.async_shutdown_send( + rstream::core::bind_handler_lifetime( + std::enable_shared_from_this::shared_from_this(), + std::move(handler))); + } + else if constexpr (requires(native_socket_type& socket, boost::system::error_code& error_code) { + socket.shutdown(boost::asio::socket_base::shutdown_send, error_code); + }) { + boost::system::error_code error_code; + m_socket.shutdown(boost::asio::socket_base::shutdown_send, error_code); + rstream::core::invoke_completion_handler(get_executor(), std::move(handler), error_code); + } + else { + rstream::core::invoke_completion_handler(get_executor(), std::move(handler), boost::asio::error::operation_not_supported); + } +} + } // namespace stream } // namespace detail } // namespace io diff --git a/lib/rstream/io/detail/stream/stream_socket_ssl.cpp b/lib/rstream/io/detail/stream/stream_socket_ssl.cpp index 72e234b..b74b1b6 100644 --- a/lib/rstream/io/detail/stream/stream_socket_ssl.cpp +++ b/lib/rstream/io/detail/stream/stream_socket_ssl.cpp @@ -123,6 +123,8 @@ class RSTREAM_GNUC_INTERNAL stream_socket_ssl::impl : public std::enable_shared_ void async_shutdown(async_shutdown_completion_handler&& handler); + void async_shutdown_send(async_shutdown_send_completion_handler&& handler); + void async_connect(const endpoint& endpoint, async_connect_completion_handler&& handler); void async_write_some(const boost::asio::const_buffer& buffer, async_write_some_completion_handler&& handler); @@ -143,6 +145,8 @@ class RSTREAM_GNUC_INTERNAL stream_socket_ssl::impl : public std::enable_shared_ void async_shutdown_internal(async_shutdown_completion_handler&& handler); + void async_shutdown_send_internal(async_shutdown_send_completion_handler&& handler); + void async_connect_internal(const endpoint& endpoint, async_connect_completion_handler&& handler); void async_write_some_internal(const boost::asio::const_buffer& buffer, async_write_some_completion_handler&& handler); @@ -154,6 +158,24 @@ class RSTREAM_GNUC_INTERNAL stream_socket_ssl::impl : public std::enable_shared_ void async_read_some_internal(const mutable_buffer_sequence_type& buffer, async_read_some_completion_handler&& handler); #endif + template + auto bind_ssl_handler(Handler&& handler) + { + auto completion_handler = rstream::core::bind_associated_handler( + std::forward(handler), + [executor = m_next_layer->get_executor()](auto& associated_handler, auto&&... args) mutable { + rstream::core::dispatch_completion_handler( + executor, + std::move(associated_handler), + std::forward(args)...); + }); +#if SSL_STREAM_USE_STRAND == 1 + return boost::asio::bind_executor(m_strand, std::move(completion_handler)); +#else + return completion_handler; +#endif + } + boost::asio::ssl::context make_ssl_context(); #if SSL_STREAM_USE_OPENSSL_ENGINE == 1 @@ -341,6 +363,11 @@ void stream_socket_ssl::async_read_some_internal(const mutable_buffer_sequence_t m_impl->async_read_some(buffer, std::move(handler)); } +void stream_socket_ssl::async_shutdown_send_internal(async_shutdown_send_completion_handler&& handler) +{ + m_impl->async_shutdown_send(std::move(handler)); +} + stream_socket_ssl::impl::impl(stream_socket_ptr next_layer, const ssl::config& config, type type, core::allocator::ptr allocator) : #if SSL_STREAM_USE_STRAND == 1 @@ -413,6 +440,13 @@ void stream_socket_ssl::impl::async_shutdown(async_shutdown_completion_handler&& } #endif +#if SSL_STREAM_USE_STRAND == 1 +void stream_socket_ssl::impl::async_shutdown_send(async_shutdown_send_completion_handler&& handler) +{ + boost::asio::dispatch(m_strand, std::bind_front(&impl::async_shutdown_send_internal, shared_from_this(), std::move(handler))); +} +#endif + #if SSL_STREAM_USE_STRAND == 1 void stream_socket_ssl::impl::async_connect(const endpoint& endpoint, async_connect_completion_handler&& handler) { @@ -480,7 +514,7 @@ void stream_socket_ssl::impl:: #endif m_ssl_stream.async_handshake( handshake_type, - rstream::core::bind_handler_lifetime(shared_from_this(), std::move(handler))); + bind_ssl_handler(rstream::core::bind_handler_lifetime(shared_from_this(), std::move(handler)))); } void stream_socket_ssl::impl:: @@ -496,8 +530,43 @@ void stream_socket_ssl::impl:: assert(m_strand.running_in_this_thread()); #endif #endif - auto operation_allocator = boost::asio::get_associated_allocator(handler); - std::allocate_shared(operation_allocator, shared_from_this(), std::forward(handler))->run(); + std::allocate_shared( + core::allocator::wrapper(m_allocator), + shared_from_this(), + std::forward(handler)) + ->run(); +} + +void stream_socket_ssl::impl:: +#if SSL_STREAM_USE_STRAND == 1 + async_shutdown_send_internal +#else + async_shutdown_send +#endif + (async_shutdown_send_completion_handler&& handler) +{ +#if SSL_STREAM_USE_STRAND == 1 +#ifdef DEBUG_BUILD + assert(m_strand.running_in_this_thread()); +#endif +#endif + auto* ssl = m_ssl_stream.native_handle(); + const auto previous_state = ::SSL_get_shutdown(ssl); + const auto previous_receive = previous_state & SSL_RECEIVED_SHUTDOWN; + if ((previous_state & SSL_SENT_SHUTDOWN) != 0) { + rstream::core::invoke_completion_handler(m_next_layer->get_executor(), std::move(handler), boost::system::error_code()); + return; + } + ::SSL_set_shutdown(ssl, previous_state | SSL_RECEIVED_SHUTDOWN); + try { + m_ssl_stream.async_shutdown(bind_ssl_handler(rstream::core::bind_handler_lifetime(shared_from_this(), std::move(handler)))); + } + catch (...) { + ::SSL_set_shutdown(ssl, previous_state); + throw; + } + const auto shutdown_state = ::SSL_get_shutdown(ssl); + ::SSL_set_shutdown(ssl, (shutdown_state & ~SSL_RECEIVED_SHUTDOWN) | previous_receive); } void stream_socket_ssl::impl:: @@ -513,8 +582,12 @@ void stream_socket_ssl::impl:: assert(m_strand.running_in_this_thread()); #endif #endif - auto operation_allocator = boost::asio::get_associated_allocator(handler); - std::allocate_shared(operation_allocator, shared_from_this(), endpoint, std::forward(handler))->run(); + std::allocate_shared( + core::allocator::wrapper(m_allocator), + shared_from_this(), + endpoint, + std::forward(handler)) + ->run(); } void stream_socket_ssl::impl:: @@ -532,7 +605,7 @@ void stream_socket_ssl::impl:: #endif m_ssl_stream.async_write_some( buffer, - rstream::core::bind_handler_lifetime(shared_from_this(), std::move(handler))); + bind_ssl_handler(rstream::core::bind_handler_lifetime(shared_from_this(), std::move(handler)))); } void stream_socket_ssl::impl:: @@ -550,7 +623,7 @@ void stream_socket_ssl::impl:: #endif m_ssl_stream.async_write_some( buffer, - rstream::core::bind_handler_lifetime(shared_from_this(), std::move(handler))); + bind_ssl_handler(rstream::core::bind_handler_lifetime(shared_from_this(), std::move(handler)))); } void stream_socket_ssl::impl:: @@ -568,7 +641,7 @@ void stream_socket_ssl::impl:: #endif m_ssl_stream.async_read_some( buffer, - rstream::core::bind_handler_lifetime(shared_from_this(), std::move(handler))); + bind_ssl_handler(rstream::core::bind_handler_lifetime(shared_from_this(), std::move(handler)))); } void stream_socket_ssl::impl:: @@ -586,7 +659,7 @@ void stream_socket_ssl::impl:: #endif m_ssl_stream.async_read_some( buffer, - rstream::core::bind_handler_lifetime(shared_from_this(), std::move(handler))); + bind_ssl_handler(rstream::core::bind_handler_lifetime(shared_from_this(), std::move(handler)))); } boost::asio::ssl::context stream_socket_ssl::impl::make_ssl_context() @@ -1733,12 +1806,22 @@ void stream_socket_ssl::impl::async_connect_operation::do_connect() on_complete(error_code); } else { - m_ptr->m_next_layer->async_connect(m_endpoint, std::bind(&async_connect_operation::on_connect, shared_from_this(), std::placeholders::_1)); + auto completion_handler = std::bind(&async_connect_operation::on_connect, shared_from_this(), std::placeholders::_1); +#if SSL_STREAM_USE_STRAND == 1 + m_ptr->m_next_layer->async_connect(m_endpoint, boost::asio::bind_executor(m_ptr->m_strand, std::move(completion_handler))); +#else + m_ptr->m_next_layer->async_connect(m_endpoint, std::move(completion_handler)); +#endif } } void stream_socket_ssl::impl::async_connect_operation::on_connect(const boost::system::error_code& error_code) { +#if SSL_STREAM_USE_STRAND == 1 +#ifdef DEBUG_BUILD + assert(m_ptr->m_strand.running_in_this_thread()); +#endif +#endif if (error_code) { on_complete(error_code); } @@ -1752,11 +1835,21 @@ void stream_socket_ssl::impl::async_connect_operation::do_handshake() #ifdef DEBUG_BUILD m_logger->trace("handshaking with SSL server..."); #endif - m_ptr->async_handshake(std::bind(&async_connect_operation::on_handshake, shared_from_this(), std::placeholders::_1)); + auto completion_handler = std::bind(&async_connect_operation::on_handshake, shared_from_this(), std::placeholders::_1); +#if SSL_STREAM_USE_STRAND == 1 + m_ptr->async_handshake(boost::asio::bind_executor(m_ptr->m_strand, std::move(completion_handler))); +#else + m_ptr->async_handshake(std::move(completion_handler)); +#endif } void stream_socket_ssl::impl::async_connect_operation::on_handshake(const boost::system::error_code& error_code) { +#if SSL_STREAM_USE_STRAND == 1 +#ifdef DEBUG_BUILD + assert(m_ptr->m_strand.running_in_this_thread()); +#endif +#endif #ifdef DEBUG_BUILD m_logger->trace("handshake with SSL server completed [error_code: {}]", (error_code ? error_code.message() : "none")); #endif diff --git a/lib/rstream/io/detail/stream/stream_socket_ssl.hpp b/lib/rstream/io/detail/stream/stream_socket_ssl.hpp index e67bf48..fde66f1 100644 --- a/lib/rstream/io/detail/stream/stream_socket_ssl.hpp +++ b/lib/rstream/io/detail/stream/stream_socket_ssl.hpp @@ -63,6 +63,8 @@ class stream_socket_ssl : public stream_socket_base { void async_read_some_internal(const mutable_buffer_sequence_type& buffer, async_read_some_completion_handler&& handler) override; + void async_shutdown_send_internal(async_shutdown_send_completion_handler&& handler) override; + std::shared_ptr m_impl; }; diff --git a/lib/rstream/io/queue.hpp b/lib/rstream/io/queue.hpp index c3c86b4..ab8840f 100644 --- a/lib/rstream/io/queue.hpp +++ b/lib/rstream/io/queue.hpp @@ -301,11 +301,10 @@ typename queue::executor_type queue::impl::get_executor() template void queue::impl::async_send(const core::buffer buffer, async_send_completion_handler&& handler) { - auto allocator = boost::asio::get_associated_allocator(handler); - auto self = std::enable_shared_from_this::shared_from_this(); - auto task_ptr = std::allocate_shared(allocator, buffer, std::move(handler)); + auto self = std::enable_shared_from_this::shared_from_this(); + auto task_ptr = std::allocate_shared(core::allocator::wrapper(m_allocator), buffer, std::move(handler)); task_ptr->arm(task_ptr, self, m_strand); - boost::asio::dispatch(m_strand, boost::asio::bind_allocator(allocator, [self, task_ptr] { self->send(task_ptr); })); + boost::asio::dispatch(m_strand, boost::asio::bind_allocator(core::allocator::wrapper(m_allocator), [self, task_ptr] { self->send(task_ptr); })); } template @@ -318,12 +317,11 @@ void queue::impl::cancel() template void queue::impl::async_cancel(async_cancel_completion_handler&& handler) { - auto allocator = boost::asio::get_associated_allocator(handler); - auto self = std::enable_shared_from_this::shared_from_this(); + auto self = std::enable_shared_from_this::shared_from_this(); boost::asio::dispatch( m_strand, boost::asio::bind_allocator( - allocator, + core::allocator::wrapper(m_allocator), [self, handler = std::move(handler)]() mutable { self->cancel_internal(std::move(handler)); })); diff --git a/lib/rstream/io/stream_socket_base.hpp b/lib/rstream/io/stream_socket_base.hpp index d828344..55b3efc 100644 --- a/lib/rstream/io/stream_socket_base.hpp +++ b/lib/rstream/io/stream_socket_base.hpp @@ -48,7 +48,29 @@ class stream_socket_base_interface { buffers); } + using async_shutdown_send_completion_handler = rstream::core::completion_handler; + template + auto async_shutdown_send(BOOST_ASIO_MOVE_ARG(shutdown_handler) handler) + { + return boost::asio::async_initiate( + async_shutdown_send_op(*this), handler); + } + private: + struct async_shutdown_send_op { + async_shutdown_send_op(stream_socket_base_interface& socket) + : m_socket(socket) + { + } + template + void operator()(BOOST_ASIO_MOVE_ARG(shutdown_handler) handler) + { + auto executor = boost::asio::get_associated_executor(handler, m_socket.m_executor); + m_socket.async_shutdown_send_internal(boost::asio::bind_executor(executor, std::forward(handler))); + } + stream_socket_base_interface& m_socket; + }; + template struct async_write_some_op { async_write_some_op(stream_socket_base_interface& socket) @@ -129,6 +151,8 @@ class stream_socket_base_interface { virtual void async_read_some_internal(const mutable_buffer_sequence_type& buffer, async_read_some_completion_handler&& handler) = 0; + virtual void async_shutdown_send_internal(async_shutdown_send_completion_handler&& handler) = 0; + io_object::executor_type m_executor; }; diff --git a/test/core/common/test_core_executor_binder.cpp b/test/core/common/test_core_executor_binder.cpp index 4363181..2fbbe00 100644 --- a/test/core/common/test_core_executor_binder.cpp +++ b/test/core/common/test_core_executor_binder.cpp @@ -180,6 +180,45 @@ static void check_completion_is_deferred_on_associated_executor() assert(allocation->m_allocations == allocation->m_deallocations); } +static void check_completion_can_dispatch_on_associated_executor() +{ + boost::asio::io_context fallback_context; + boost::asio::io_context associated_context; + auto strand = boost::asio::make_strand(associated_context); + auto allocation = std::make_shared(); + boost::asio::cancellation_signal cancellation; + std::atomic_size_t calls = 0; + auto work = boost::asio::make_work_guard(associated_context); + boost::asio::post(strand, [&] { + rstream::core::completion_handler handler( + associated_handler(strand, counting_allocator(allocation), cancellation.slot(), calls)); + rstream::core::dispatch_completion_handler(fallback_context.get_executor(), std::move(handler), boost::system::error_code()); + assert(calls == 1); + work.reset(); + }); + associated_context.run(); + assert(calls == 1); + assert(allocation->m_allocations == allocation->m_deallocations); +} + +static void check_abandoned_completion_releases_associated_state() +{ + auto allocation = std::make_shared(); + std::atomic_size_t calls = 0; + { + boost::asio::io_context associated_context; + boost::asio::io_context fallback_context; + auto strand = boost::asio::make_strand(associated_context); + boost::asio::cancellation_signal cancellation; + rstream::core::completion_handler handler( + associated_handler(strand, counting_allocator(allocation), cancellation.slot(), calls)); + rstream::core::invoke_completion_handler(fallback_context.get_executor(), std::move(handler), boost::system::error_code()); + } + assert(calls == 0); + assert(allocation->m_allocations > 0); + assert(allocation->m_allocations == allocation->m_deallocations); +} + static void check_adapter_preserves_associations() { boost::asio::io_context fallback_context; @@ -320,6 +359,8 @@ int main() { check_type_erasure_preserves_associations(); check_completion_is_deferred_on_associated_executor(); + check_completion_can_dispatch_on_associated_executor(); + check_abandoned_completion_releases_associated_state(); check_adapter_preserves_associations(); check_lifetime_adapter_preserves_owner_and_associations(); check_strand_serializes_type_erased_handlers(); diff --git a/test/io/common/test_io_common_queue.cpp b/test/io/common/test_io_common_queue.cpp index 824e730..81f50d4 100644 --- a/test/io/common/test_io_common_queue.cpp +++ b/test/io/common/test_io_common_queue.cpp @@ -15,6 +15,7 @@ #include #include #include +#include #include #include #include @@ -26,6 +27,54 @@ #include #include +struct allocator_state { + std::atomic_size_t m_allocations = 0; + std::atomic_size_t m_deallocations = 0; +}; + +template +class stateful_allocator { + public: + using value_type = T; + + explicit stateful_allocator(std::shared_ptr state) + : m_state(std::move(state)) + { + } + + template + stateful_allocator(const stateful_allocator& other) + : m_state(other.state()) + { + } + + T* allocate(std::size_t count) + { + ++m_state->m_allocations; + return std::allocator().allocate(count); + } + + void deallocate(T* pointer, std::size_t count) + { + ++m_state->m_deallocations; + std::allocator().deallocate(pointer, count); + } + + const std::shared_ptr& state() const + { + return m_state; + } + + template + bool operator==(const stateful_allocator& other) const + { + return m_state == other.state(); + } + + private: + std::shared_ptr m_state; +}; + class controlled_transport { public: using executor_type = boost::asio::io_context::executor_type; @@ -446,6 +495,23 @@ static void check_deferred_operations_are_lazy() assert(cancel_calls == 1); } +static void check_abandoned_type_erased_operations_are_safe() +{ + auto allocation = std::make_shared(); + { + boost::asio::io_context io_context; + controlled_transport transport(io_context.get_executor()); + rstream::io::queue queue(transport); + queue.async_send( + make_buffer(1), + boost::asio::bind_allocator(stateful_allocator(allocation), [](const boost::system::error_code&) {})); + queue.async_cancel( + boost::asio::bind_allocator(stateful_allocator(allocation), [](const boost::system::error_code&) {})); + } + assert(allocation->m_allocations > 0); + assert(allocation->m_allocations == allocation->m_deallocations); +} + int main() { check_owned_move_only_transport(); @@ -457,5 +523,6 @@ int main() check_async_cancel_waits_for_active_send(); check_async_cancel_supports_multiple_waiters(); check_deferred_operations_are_lazy(); + check_abandoned_type_erased_operations_are_safe(); return 0; } diff --git a/test/io/common/test_io_common_stream_tls.cpp b/test/io/common/test_io_common_stream_tls.cpp index 8ed7d21..35e4c7d 100644 --- a/test/io/common/test_io_common_stream_tls.cpp +++ b/test/io/common/test_io_common_stream_tls.cpp @@ -5,10 +5,12 @@ #include #include #include +#include #include #include #include #include +#include #include #include #include @@ -16,6 +18,7 @@ #include #include +#include #include #include #include @@ -200,6 +203,11 @@ class fake_stream_socket : public rstream::io::detail::stream::stream_socket_int rstream::core::invoke_completion_handler(get_executor(), std::move(handler), boost::asio::error::would_block, boost::asio::buffer_size(buffer)); } + void async_shutdown_send_internal(async_shutdown_send_completion_handler&& handler) override + { + rstream::core::invoke_completion_handler(get_executor(), std::move(handler), boost::system::error_code()); + } + endpoint_type m_endpoint; }; @@ -636,6 +644,224 @@ static void check_tls_accept_connect_and_transfer(const std::string& groups_quer assert(server_sequence_sent); } +class tls_half_close_exchange : public std::enable_shared_from_this { + public: + tls_half_close_exchange(rstream::io::stream::stream_socket& client, rstream::io::stream::stream_socket& server) + : m_client(client), + m_server(server), + m_payload(1024 * 1024, '\0'), + m_response("response-after-client-eof"), + m_client_response(m_response.size(), '\0') + { + for (std::size_t i = 0; i < m_payload.size(); ++i) { + m_payload[i] = static_cast((i * 31U + 17U) % 251U); + } + } + + void run() + { + read_server(); + boost::asio::async_read( + m_client, + boost::asio::buffer(m_client_response), + [self = shared_from_this()](const boost::system::error_code& error_code, std::size_t size) { + self->m_client_read_error = error_code; + self->m_client_read_size = size; + self->m_client_read_done = true; + self->mark_completed(g_client_read_completed); + }); + boost::asio::async_write( + m_client, + boost::asio::buffer(m_payload), + [self = shared_from_this()](const boost::system::error_code& error_code, std::size_t size) { + self->m_client_write_error = error_code; + self->m_client_write_size = size; + self->m_client_write_done = true; + self->mark_completed(g_client_write_completed); + self->m_client.async_shutdown_send( + [self](const boost::system::error_code& shutdown_error) { + self->on_shutdown_send(shutdown_error); + }); + }); + } + + bool completed() const + { + return m_completion_mask.load(std::memory_order_acquire) == g_all_completed; + } + + bool wait_for_completion(std::chrono::milliseconds timeout) + { + std::unique_lock lock(m_completion_mutex); + return m_completion_condition.wait_for(lock, timeout, [this] { return completed(); }); + } + + void assert_success() const + { + assert(!m_client_write_error); + assert(m_client_write_size == m_payload.size()); + assert(m_server_read_error == boost::asio::error::eof); + assert(m_server_payload == m_payload); + assert(!m_server_write_error); + assert(m_server_write_size == m_response.size()); + assert(!m_client_read_error); + assert(m_client_read_size == m_response.size()); + assert(m_client_response == m_response); + assert(!m_first_shutdown_error); + assert(!m_second_shutdown_error); + assert(m_shutdown_count == 2); + } + + private: + void read_server() + { + m_server.async_read_some( + boost::asio::buffer(m_server_buffer), + [self = shared_from_this()](const boost::system::error_code& error_code, std::size_t size) { + self->m_server_payload.append(self->m_server_buffer.data(), size); + if (!error_code) { + self->read_server(); + return; + } + self->m_server_read_error = error_code; + self->m_server_read_done = true; + self->mark_completed(g_server_read_completed); + if (error_code == boost::asio::error::eof) { + boost::asio::async_write( + self->m_server, + boost::asio::buffer(self->m_response), + [self](const boost::system::error_code& write_error, std::size_t write_size) { + self->m_server_write_error = write_error; + self->m_server_write_size = write_size; + self->m_server_write_done = true; + self->mark_completed(g_server_write_completed); + }); + } + else { + self->m_server_write_done = true; + self->mark_completed(g_server_write_completed); + } + }); + } + + void on_shutdown_send(const boost::system::error_code& error_code) + { + ++m_shutdown_count; + if (m_shutdown_count == 1) { + m_first_shutdown_error = error_code; + m_client.async_shutdown_send( + [self = shared_from_this()](const boost::system::error_code& repeated_error) { + self->on_shutdown_send(repeated_error); + }); + } + else { + m_second_shutdown_error = error_code; + mark_completed(g_shutdown_completed); + } + } + + void mark_completed(unsigned int bit) + { + const auto previous = m_completion_mask.fetch_or(bit, std::memory_order_acq_rel); + assert((previous & bit) == 0); + if ((previous | bit) == g_all_completed) { + std::lock_guard lock(m_completion_mutex); + m_completion_condition.notify_all(); + } + } + + static constexpr unsigned int g_client_write_completed = 1U << 0U; + static constexpr unsigned int g_shutdown_completed = 1U << 1U; + static constexpr unsigned int g_server_read_completed = 1U << 2U; + static constexpr unsigned int g_server_write_completed = 1U << 3U; + static constexpr unsigned int g_client_read_completed = 1U << 4U; + static constexpr unsigned int g_all_completed = g_client_write_completed | g_shutdown_completed | g_server_read_completed | g_server_write_completed | g_client_read_completed; + + rstream::io::stream::stream_socket& m_client; + rstream::io::stream::stream_socket& m_server; + std::string m_payload; + std::string m_response; + std::string m_client_response; + std::string m_server_payload; + std::array m_server_buffer{}; + boost::system::error_code m_client_write_error; + boost::system::error_code m_server_read_error; + boost::system::error_code m_server_write_error; + boost::system::error_code m_client_read_error; + boost::system::error_code m_first_shutdown_error; + boost::system::error_code m_second_shutdown_error; + std::size_t m_client_write_size = 0; + std::size_t m_server_write_size = 0; + std::size_t m_client_read_size = 0; + unsigned int m_shutdown_count = 0; + std::atomic_uint m_completion_mask = 0; + std::mutex m_completion_mutex; + std::condition_variable m_completion_condition; + bool m_client_write_done = false; + bool m_server_read_done = false; + bool m_server_write_done = false; + bool m_client_read_done = false; +}; + +static void check_tls_half_close_preserves_receive_direction() +{ + certificate_files files; + boost::asio::io_context io_context; + const auto port = unused_tcp_port(); + const auto server_endpoint = resolve_one( + io_context, + "tcp://127.0.0.1:" + std::to_string(port) + "?ssl&ssl.cert_file=" + files.cert_file() + + "&ssl.key_file=" + files.key_file() + + "&ssl.peer_verification=false&ssl.request_peer_cert=false&ssl.async_shutdown_timeout_ms=0"); + const auto client_endpoint = resolve_one( + io_context, + "tcp://127.0.0.1:" + std::to_string(port) + + "?ssl&ssl.peer_verification=false&ssl.request_peer_cert=false&ssl.sni=localhost&ssl.async_shutdown_timeout_ms=0"); + rstream::io::stream::acceptor acceptor(io_context.get_executor()); + boost::system::error_code error_code; + acceptor.open(server_endpoint, error_code); + assert(!error_code); + acceptor.bind(server_endpoint, error_code); + assert(!error_code); + acceptor.listen(boost::asio::socket_base::max_listen_connections, error_code); + assert(!error_code); + rstream::io::stream::stream_socket server(io_context.get_executor()); + rstream::io::stream::stream_socket client(io_context.get_executor()); + rstream::io::stream::endpoint remote_endpoint; + bool accepted = false; + bool connected = false; + acceptor.async_accept(server, remote_endpoint, [&](const boost::system::error_code& error) { + assert(!error); + accepted = true; + }); + client.async_connect(client_endpoint, [&](const boost::system::error_code& error) { + assert(!error); + connected = true; + }); + run_until(io_context, [&] { return accepted && connected; }); + auto exchange = std::make_shared(client, server); + exchange->run(); + auto work = boost::asio::make_work_guard(io_context); + std::array workers; + for (auto& worker : workers) { + worker = std::thread([&] { io_context.run(); }); + } + const auto completed = exchange->wait_for_completion(std::chrono::seconds(10)); + boost::system::error_code ignored; + client.close(ignored); + server.close(ignored); + acceptor.close(ignored); + work.reset(); + if (!completed) { + io_context.stop(); + } + for (auto& worker : workers) { + worker.join(); + } + assert(completed); + exchange->assert_success(); +} + static void check_tls_shutdown_timeout_is_serialized() { certificate_files files; @@ -825,6 +1051,7 @@ int main(int argc, char** argv) check_tls_config_errors(); check_direct_tls_context_configuration(); check_tls_accept_connect_and_transfer(); + check_tls_half_close_preserves_receive_direction(); check_tls_shutdown_timeout_is_serialized(); check_tls_accept_preserves_peer_executor(); check_tls_peer_verification_checks_hostname(); diff --git a/test/io/io-rstrm/test_io_rstrm_handshake.cpp b/test/io/io-rstrm/test_io_rstrm_handshake.cpp index dabe241..8f00b58 100644 --- a/test/io/io-rstrm/test_io_rstrm_handshake.cpp +++ b/test/io/io-rstrm/test_io_rstrm_handshake.cpp @@ -103,7 +103,11 @@ static void check_zero_rtt_stream_request_does_not_wait_for_response() rstream::test::connect_stream_pair(*socket_a, *socket_b); test_stream stream(*socket_a, true); rstream::io_rstrm::config config; +#ifdef RSTREAM_WITH_IO_STREAMS config.m_token = "secret-token"; +#else + config.m_no_token = true; +#endif config.m_zero_rtt = true; bool client_called = false; bool server_called = false; @@ -123,8 +127,12 @@ static void check_zero_rtt_stream_request_does_not_wait_for_response() assert(message.stream_req().tunnel_id_name() == "api"); assert(message.stream_req().has_zero_rtt()); assert(message.stream_req().zero_rtt().value()); +#ifdef RSTREAM_WITH_IO_STREAMS assert(message.stream_req().client_details().has_token()); assert(message.stream_req().client_details().token().value() == "secret-token"); +#else + assert(!message.stream_req().client_details().has_token()); +#endif server_called = true; socket->close(); co_return; }, boost::asio::detached); @@ -256,6 +264,7 @@ static void check_proxy_success_response_completes() assert(server_called); } +#ifdef RSTREAM_WITH_IO_STREAMS static void check_proxy_secret_is_allowed_with_mtls_agent_auth() { boost::asio::io_context io_context; @@ -293,6 +302,7 @@ static void check_proxy_secret_is_allowed_with_mtls_agent_auth() assert(client_called); assert(server_called); } +#endif static void check_unexpected_response_type_is_rejected() { @@ -429,7 +439,9 @@ int main(int argc, char** argv) check_stream_response_error_is_mapped(); check_stream_success_response_completes(); check_proxy_success_response_completes(); +#ifdef RSTREAM_WITH_IO_STREAMS check_proxy_secret_is_allowed_with_mtls_agent_auth(); +#endif check_unexpected_response_type_is_rejected(); check_invalid_protobuf_response_is_rejected(); check_cancellation_reaches_the_transport(); diff --git a/test/ncat/CMakeLists.txt b/test/ncat/CMakeLists.txt index cc0d3c0..2a9290b 100644 --- a/test/ncat/CMakeLists.txt +++ b/test/ncat/CMakeLists.txt @@ -20,4 +20,9 @@ if(WIN32) else() add_test_target(${PROJECT_NAME}-test-ncat-runtime test_ncat_runtime.cpp ${PROJECT_NAME}::ncat ${PROJECT_NAME}::${PROJECT_NAME}) rstream_enable_runtime_plugins(${PROJECT_NAME}-test-ncat-runtime ${PROJECT_NAME}-plugin-io-generic) + add_test( + NAME ${PROJECT_NAME}-test-ncat-cli-runtime + COMMAND ${Python3_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/test_ncat_cli_runtime.py $) + rstream_configure_test(${PROJECT_NAME}-test-ncat-cli-runtime) + set_property(GLOBAL APPEND PROPERTY RSTREAM_TEST_TARGETS ${PROJECT_NAME}-ncat) endif() diff --git a/test/ncat/test_ncat_cli_runtime.py b/test/ncat/test_ncat_cli_runtime.py new file mode 100644 index 0000000..9ce4066 --- /dev/null +++ b/test/ncat/test_ncat_cli_runtime.py @@ -0,0 +1,66 @@ +#!/usr/bin/env python3 + +import signal +import socket +import subprocess +import sys +import time + + +def allocate_address() -> str: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as listener: + listener.bind(("127.0.0.1", 0)) + return f"127.0.0.1:{listener.getsockname()[1]}" + + +def run_case(binary: str, command_arguments: list[str], expected: bytes) -> None: + address = allocate_address() + server = subprocess.Popen( + [binary, "-L", address, *command_arguments, "--jobs=2"], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + try: + deadline = time.monotonic() + 10 + while True: + client = subprocess.run( + [binary, address, "-I", "--jobs=2"], + capture_output=True, + timeout=10, + check=False, + ) + if client.returncode == 0: + if client.stdout != expected: + raise AssertionError( + f"unexpected output {client.stdout!r}, expected {expected!r}; " + f"client stderr={client.stderr!r}" + ) + break + if server.poll() is not None: + stdout, stderr = server.communicate() + raise AssertionError( + f"server exited with {server.returncode}; stdout={stdout!r}; stderr={stderr!r}" + ) + if time.monotonic() >= deadline: + raise AssertionError(f"client did not connect; stderr={client.stderr!r}") + time.sleep(0.05) + finally: + if server.poll() is None: + server.send_signal(signal.SIGTERM) + try: + server.communicate(timeout=10) + except subprocess.TimeoutExpired: + server.kill() + server.communicate() + + +def main() -> None: + if len(sys.argv) != 2: + raise SystemExit("usage: test_ncat_cli_runtime.py ") + command = "exec printf cli-shell-ok" + run_case(sys.argv[1], ["-c", command], b"cli-shell-ok") + run_case(sys.argv[1], [f"-c={command}"], b"cli-shell-ok") + + +if __name__ == "__main__": + main() diff --git a/test/tunnel/test_tunnel_proxy.cpp b/test/tunnel/test_tunnel_proxy.cpp index abe490d..e53e690 100644 --- a/test/tunnel/test_tunnel_proxy.cpp +++ b/test/tunnel/test_tunnel_proxy.cpp @@ -222,9 +222,10 @@ class tcp_thread_server { class fake_engine { public: - explicit fake_engine(std::atomic_bool& stream_exchanged) + explicit fake_engine(std::atomic_bool& stream_exchanged, bool half_close = false) : m_acceptor(m_io_context, tcp::endpoint(boost::asio::ip::make_address("127.0.0.1"), 0)), - m_stream_exchanged(stream_exchanged) + m_stream_exchanged(stream_exchanged), + m_half_close(half_close) { } @@ -272,6 +273,11 @@ class fake_engine { const std::string downstream_payload = "hello"; boost::asio::write(stream, boost::asio::buffer(downstream_payload)); + if (m_half_close) { + boost::system::error_code error_code; + stream.shutdown(tcp::socket::shutdown_send, error_code); + check(!error_code, "failed to half-close downstream test stream"); + } std::array upstream_reply{}; boost::asio::read(stream, boost::asio::buffer(upstream_reply)); check(std::string(upstream_reply.data(), upstream_reply.size()) == "world", "unexpected proxied reply"); @@ -301,25 +307,37 @@ class fake_engine { boost::asio::io_context m_io_context; tcp::acceptor m_acceptor; std::atomic_bool& m_stream_exchanged; + bool m_half_close; std::thread m_thread; std::exception_ptr m_exception; }; -static void check_proxy_forwards_engine_stream_to_upstream_and_back() +static void check_proxy_forwards_engine_stream_to_upstream_and_back(bool half_close = false) { std::atomic_bool upstream_served = false; - tcp_thread_server upstream([&](tcp::socket& socket) { + tcp_thread_server upstream([&, half_close](tcp::socket& socket) { std::array request{}; boost::asio::read(socket, boost::asio::buffer(request)); check(std::string(request.data(), request.size()) == "hello", "unexpected upstream request"); + if (half_close) { + std::array trailing{}; + boost::system::error_code error_code; + socket.read_some(boost::asio::buffer(trailing), error_code); + check(error_code == boost::asio::error::eof, "upstream did not receive the downstream half-close"); + } const std::string response = "world"; boost::asio::write(socket, boost::asio::buffer(response)); + if (half_close) { + boost::system::error_code error_code; + socket.shutdown(tcp::socket::shutdown_send, error_code); + check(!error_code, "failed to half-close upstream test stream"); + } upstream_served = true; }); upstream.start(); std::atomic_bool stream_exchanged = false; - fake_engine engine(stream_exchanged); + fake_engine engine(stream_exchanged, half_close); engine.start(); boost::asio::io_context io_context; @@ -500,6 +518,7 @@ int main(int argc, char** argv) (void)argc; (void)argv; check_proxy_forwards_engine_stream_to_upstream_and_back(); + check_proxy_forwards_engine_stream_to_upstream_and_back(true); check_proxy_rejects_second_run_while_active(); check_proxy_default_tunnel_request_leaves_public_policy_to_server(); return 0; From 9a494668d766a63ab875e3418a53815c527917dd Mon Sep 17 00:00:00 2001 From: uartnet <140632163+uartnet@users.noreply.github.com> Date: Sat, 15 Aug 2026 01:17:27 +0200 Subject: [PATCH 2/9] feat(runtime): negotiate bounded control liveness --- lib/rstream/io-rstrm/client.cpp | 120 +++++- .../rstream/io-rstrm/protobuf/messages.proto | 14 +- .../test_io_rstrm_control_channel.cpp | 404 +++++++++++++++++- 3 files changed, 521 insertions(+), 17 deletions(-) diff --git a/lib/rstream/io-rstrm/client.cpp b/lib/rstream/io-rstrm/client.cpp index 401487a..ae42663 100644 --- a/lib/rstream/io-rstrm/client.cpp +++ b/lib/rstream/io-rstrm/client.cpp @@ -52,6 +52,10 @@ namespace rstream { namespace io_rstrm { +static constexpr unsigned int kMinControlHeartbeatIntervalMs = 1000; +static constexpr unsigned int kMaxControlHeartbeatIntervalMs = 300000; +static constexpr unsigned int kMaxControlHeartbeatTimeoutMs = 900000; + static bool invalid_published_tcp_options(const tunnel_properties& properties) { const bool published_tcp = properties.m_protocol && properties.m_protocol.value() == protocol::tcp; @@ -245,6 +249,8 @@ class RSTREAM_GNUC_INTERNAL client::impl : public std::enable_shared_from_this effective_client_details() const; void async_connect_internal(const connect_op_type::ptr& op); @@ -359,6 +365,8 @@ class RSTREAM_GNUC_INTERNAL client::impl : public std::enable_shared_from_this(operation_allocator, std::move(handler)); + const auto op = std::allocate_shared(core::allocator::wrapper(m_allocator), std::move(handler)); { std::lock_guard lock(m_mutex); if (!m_is_state_non_null) { @@ -668,9 +685,8 @@ void client::impl::async_connect(async_connect_completion_handler&& handler) void client::impl::async_create_tunnel(const tunnel_properties& properties, async_create_tunnel_completion_handler&& handler) { - auto operation_allocator = boost::asio::get_associated_allocator(handler); - const auto op = std::allocate_shared( - operation_allocator, + const auto op = std::allocate_shared( + core::allocator::wrapper(m_allocator), normalize_tunnel_properties(properties), std::move(handler)); auto cancellation_slot = boost::asio::get_associated_cancellation_slot(op->m_handler); @@ -697,9 +713,8 @@ void client::impl::async_create_tunnel(const tunnel_properties& properties, asyn void client::impl::async_accept_tunnel(const std::string& tunnel_id, socket& peer, endpoint& endpoint, tunnel::async_accept_completion_handler&& handler) { - auto operation_allocator = boost::asio::get_associated_allocator(handler); - const auto op = std::allocate_shared( - operation_allocator, + const auto op = std::allocate_shared( + core::allocator::wrapper(m_allocator), peer, endpoint, std::move(handler)); @@ -806,6 +821,24 @@ void client::impl::arm_state_timer(unsigned int timeout_ms) task_ptr->m_timer.async_wait(completion_handler); } +void client::impl::arm_liveness_timer() +{ +#ifdef DEBUG_BUILD + assert(m_strand.running_in_this_thread()); +#endif + if (m_heartbeat_timeout_ms == 0 || m_state != state::connected) { + return; + } + const auto generation = m_generation; + m_liveness_timer.expires_after(std::chrono::milliseconds(m_heartbeat_timeout_ms)); + auto self = shared_from_this(); + m_liveness_timer.async_wait(boost::asio::bind_executor(m_strand, [self, generation](const boost::system::error_code& error_code) { + if (!error_code && generation == self->m_generation && self->m_state == state::connected) { + self->on_error(error::code::operation_timeout); + } + })); +} + void client::impl::complete_connect(const connect_op_type::ptr& op, const boost::system::error_code& error_code) { if (!op || op->m_completed) { @@ -1166,6 +1199,16 @@ void client::impl::do_open() } if (!error_code) { payload.mutable_client_details()->CopyFrom(proto_client_details); + if (m_config.m_hearbeat) { + if (m_config.m_heartbeat_interval_ms < kMinControlHeartbeatIntervalMs || m_config.m_heartbeat_interval_ms > kMaxControlHeartbeatIntervalMs) { + error_code = error::code::invalid_configuration; + } + else { + payload.mutable_liveness()->set_heartbeat_interval_ms(m_config.m_heartbeat_interval_ms); + } + } + } + if (!error_code) { message.mutable_open_control_channel_req()->CopyFrom(payload); } } @@ -1519,7 +1562,22 @@ void client::impl::on_read_incoming_message(generation_type generation, const pr if (payload.has_ok()) { const auto& ok = payload.ok(); if (!ok.client_id().empty()) { - if (ok.has_server_details()) { + m_heartbeat_timeout_ms = 0; + m_heartbeat_sequence = 0; + m_heartbeat_acknowledgement = 0; + if (ok.has_liveness()) { + const auto& liveness = ok.liveness(); + if (!m_config.m_hearbeat + || liveness.heartbeat_interval_ms() != m_config.m_heartbeat_interval_ms + || liveness.heartbeat_timeout_ms() < liveness.heartbeat_interval_ms() + || liveness.heartbeat_timeout_ms() > kMaxControlHeartbeatTimeoutMs) { + error_code = error::code::protocol_error; + } + else { + m_heartbeat_timeout_ms = liveness.heartbeat_timeout_ms(); + } + } + if (!error_code && ok.has_server_details()) { const auto& details = ok.server_details(); if (details.has_plan()) { m_status.m_plan = details.plan().value(); @@ -1534,7 +1592,9 @@ void client::impl::on_read_incoming_message(generation_type generation, const pr m_status.m_update = details.update().value(); } } - on_open(); + if (!error_code) { + on_open(); + } } else { #ifdef DEBUG_BUILD @@ -1676,6 +1736,20 @@ void client::impl::on_read_incoming_message(generation_type generation, const pr } } } + else if (message_type == payload_type::kHeartbeat) { + const auto& heartbeat = message.heartbeat(); + if (m_heartbeat_timeout_ms == 0) { + if (heartbeat.sequence() != 0 || heartbeat.acknowledgement() != 0) { + error_code = error::code::protocol_error; + } + } + else if (heartbeat.sequence() != 0 || heartbeat.acknowledgement() == 0 || heartbeat.acknowledgement() <= m_heartbeat_acknowledgement || heartbeat.acknowledgement() > m_heartbeat_sequence) { + error_code = error::code::protocol_error; + } + else { + m_heartbeat_acknowledgement = heartbeat.acknowledgement(); + } + } else if (message_type == payload_type::kCloseControlChannelRsp) { boost::system::error_code error; if (m_state != state::closing) { @@ -1691,6 +1765,9 @@ void client::impl::on_read_incoming_message(generation_type generation, const pr on_error(error_code); } else { + if (m_state == state::connected) { + arm_liveness_timer(); + } do_read_incoming_message(); } } @@ -1746,6 +1823,7 @@ void client::impl::close_internal(const boost::system::error_code& error_code) #ifdef DEBUG_BUILD m_logger->trace("closing client..."); #endif + m_liveness_timer.cancel(); set_state(state::closing); arm_state_timer(m_config.m_connection_timeout_ms); if (error_code && !m_error_code) { @@ -1791,6 +1869,7 @@ void client::impl::on_close(const boost::system::error_code& error_code) disconnection_callback = m_control_callbacks.m_on_disconnection_cb; } m_close_pending = true; + m_liveness_timer.cancel(); ++m_generation; if (m_state != state::closing) { set_state(state::closing); @@ -1828,6 +1907,9 @@ void client::impl::on_queue_cancelled( m_error_code = boost::system::error_code(); m_client_details = {}; m_status = {}; + m_heartbeat_timeout_ms = 0; + m_heartbeat_sequence = 0; + m_heartbeat_acknowledgement = 0; set_state(state::null); m_close_pending = false; { @@ -1859,7 +1941,19 @@ void client::impl::send_heartbeat() #ifdef DEBUG_BUILD assert(m_strand.running_in_this_thread()); #endif - do_send_heartbeat(); + if (m_config.m_heartbeat_interval_ms == 0 || m_state != state::connected) { + return; + } + protobuf::Message message; + auto* heartbeat = message.mutable_heartbeat(); + if (m_heartbeat_timeout_ms > 0) { + ++m_heartbeat_sequence; + if (m_heartbeat_sequence == 0) { + ++m_heartbeat_sequence; + } + heartbeat->set_sequence(m_heartbeat_sequence); + } + do_send_message(message, std::bind(&impl::do_send_heartbeat, shared_from_this())); } void client::impl::do_send_heartbeat() @@ -1905,9 +1999,7 @@ void client::impl::do_send_heartbeat() return; } if (!error_code && generation == ptr->m_generation) { - protobuf::Message message; - message.mutable_heartbeat(); - ptr->do_send_message(message, std::bind(&impl::do_send_heartbeat, ptr)); + ptr->send_heartbeat(); } task_ptr->clean(); }; diff --git a/lib/rstream/io-rstrm/definitions/rstream/io-rstrm/protobuf/messages.proto b/lib/rstream/io-rstrm/definitions/rstream/io-rstrm/protobuf/messages.proto index 7d76ede..43f24c4 100644 --- a/lib/rstream/io-rstrm/definitions/rstream/io-rstrm/protobuf/messages.proto +++ b/lib/rstream/io-rstrm/definitions/rstream/io-rstrm/protobuf/messages.proto @@ -12,7 +12,7 @@ extend google.protobuf.FieldOptions { string access = 51234; } -option (protocol_version) = "1.4.4"; +option (protocol_version) = "1.4.5"; package rstream.io_rstrm.protobuf; @@ -134,6 +134,7 @@ message TunnelProperties { message OpenControlChannelReq { ClientDetails client_details = 1; + ControlChannelLiveness liveness = 2; } // The server responds to an 'OpenControlChannelReq' message with an @@ -146,6 +147,7 @@ message OpenControlChannelRsp { message Ok { string client_id = 1; ServerDetails server_details = 2; + ControlChannelLiveness liveness = 3; } oneof payload { Ok ok = 1; @@ -261,7 +263,15 @@ message DatagramChannelClose { // Sent by client and/or server to maintain the control channel active -message Heartbeat { } +message ControlChannelLiveness { + uint32 heartbeat_interval_ms = 1; + uint32 heartbeat_timeout_ms = 2; +} + +message Heartbeat { + uint64 sequence = 1; + uint64 acknowledgement = 2; +} // Allows the server to send unsolicited messages to the client message ServerMessage { diff --git a/test/io/io-rstrm/test_io_rstrm_control_channel.cpp b/test/io/io-rstrm/test_io_rstrm_control_channel.cpp index 4e79635..7779c72 100644 --- a/test/io/io-rstrm/test_io_rstrm_control_channel.cpp +++ b/test/io/io-rstrm/test_io_rstrm_control_channel.cpp @@ -191,6 +191,15 @@ static protobuf::Message open_control_response() return response; } +static protobuf::Message open_control_liveness_response(std::uint32_t heartbeat_interval_ms, std::uint32_t heartbeat_timeout_ms) +{ + auto response = open_control_response(); + auto* liveness = response.mutable_open_control_channel_rsp()->mutable_ok()->mutable_liveness(); + liveness->set_heartbeat_interval_ms(heartbeat_interval_ms); + liveness->set_heartbeat_timeout_ms(heartbeat_timeout_ms); + return response; +} + static protobuf::Message open_tunnel_response(const protobuf::OpenTunnelReq& request) { protobuf::Message response; @@ -1096,6 +1105,8 @@ static void check_client_heartbeat_and_unexpected_close_response() auto heartbeat = read_message(socket); assert(heartbeat.has_heartbeat()); + assert(heartbeat.heartbeat().sequence() == 0); + assert(heartbeat.heartbeat().acknowledgement() == 0); write_message(socket, close_control_response()); wait_for_peer_close(socket); }); @@ -1104,7 +1115,7 @@ static void check_client_heartbeat_and_unexpected_close_response() rstream::io_rstrm::config_client config; config.m_no_token = true; config.m_hearbeat = true; - config.m_heartbeat_interval_ms = 1; + config.m_heartbeat_interval_ms = 1000; config.m_connection_timeout_ms = kControlChannelTimeoutMs; rstream::io_rstrm::client client(io_context.get_executor(), config); @@ -1135,6 +1146,325 @@ static void check_client_heartbeat_and_unexpected_close_response() assert(disconnected); } +static void check_client_negotiates_liveness_and_accepts_delayed_acknowledgement() +{ + fake_engine engine; + engine.start([](tcp::socket& socket) { + const auto open_request = read_message(socket); + check(open_request.has_open_control_channel_req(), "missing control channel request"); + const auto& request = open_request.open_control_channel_req(); + check(request.has_liveness(), "control channel request did not advertise liveness"); + check(request.liveness().heartbeat_interval_ms() == 1000, "control channel request advertised the wrong heartbeat interval"); + check(request.liveness().heartbeat_timeout_ms() == 0, "control channel request selected its own timeout"); + write_message(socket, open_control_liveness_response(1000, 1000)); + const auto heartbeat = read_message(socket); + check(heartbeat.has_heartbeat(), "client did not send a negotiated heartbeat"); + check(heartbeat.heartbeat().sequence() == 1, "negotiated heartbeat sequence is invalid"); + check(heartbeat.heartbeat().acknowledgement() == 0, "client sent a heartbeat acknowledgement"); + std::this_thread::sleep_for(std::chrono::milliseconds(800)); + protobuf::Message acknowledgement; + acknowledgement.mutable_heartbeat()->set_acknowledgement(heartbeat.heartbeat().sequence()); + write_message(socket, acknowledgement); + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + write_message(socket, close_control_response()); + wait_for_peer_close(socket); + }); + boost::asio::io_context io_context; + rstream::io_rstrm::config_client config; + config.m_no_token = true; + config.m_hearbeat = true; + config.m_heartbeat_interval_ms = 1000; + config.m_connection_timeout_ms = kControlChannelTimeoutMs; + rstream::io_rstrm::client client(io_context.get_executor(), config); + bool connected = false; + bool disconnected = false; + watchdog test_watchdog(io_context); + rstream::io_rstrm::client::control_callbacks callbacks; + callbacks.m_on_disconnection_cb = [&](const boost::system::error_code& error_code) { + check(error_code == rstream::io_rstrm::error::code::server_error, "delayed heartbeat acknowledgement did not preserve the connection"); + disconnected = true; + }; + boost::system::error_code callback_error; + client.set_control_callbacks(callbacks, callback_error); + check(!callback_error, "failed to install liveness callback"); + client.async_connect(rstream::io::make_address(engine.address()), [&](const boost::system::error_code& error_code) { + check(!error_code, "client failed to connect with negotiated liveness"); + connected = true; + }); + io_context.run(); + test_watchdog.complete(); + engine.join(); + check(!test_watchdog.timed_out(), "negotiated liveness check timed out"); + check(connected, "client did not connect with negotiated liveness"); + check(disconnected, "client did not process the post-acknowledgement close response"); +} + +static void check_client_expires_missing_liveness_acknowledgement() +{ + fake_engine engine; + engine.start([](tcp::socket& socket) { + const auto open_request = read_message(socket); + check(open_request.has_open_control_channel_req(), "missing control channel request"); + check(open_request.open_control_channel_req().has_liveness(), "control channel request did not advertise liveness"); + write_message(socket, open_control_liveness_response(1000, 1000)); + const auto heartbeat = read_message(socket); + check(heartbeat.has_heartbeat() && heartbeat.heartbeat().sequence() == 1, "client did not send the first negotiated heartbeat"); + wait_for_peer_close(socket); + }); + boost::asio::io_context io_context; + rstream::io_rstrm::config_client config; + config.m_no_token = true; + config.m_hearbeat = true; + config.m_heartbeat_interval_ms = 1000; + config.m_connection_timeout_ms = kControlChannelTimeoutMs; + rstream::io_rstrm::client client(io_context.get_executor(), config); + bool connected = false; + bool disconnected = false; + watchdog test_watchdog(io_context); + rstream::io_rstrm::client::control_callbacks callbacks; + callbacks.m_on_disconnection_cb = [&](const boost::system::error_code& error_code) { + check(error_code == rstream::io_rstrm::error::code::operation_timeout, "missing heartbeat acknowledgement returned the wrong error"); + disconnected = true; + }; + boost::system::error_code callback_error; + client.set_control_callbacks(callbacks, callback_error); + check(!callback_error, "failed to install liveness callback"); + client.async_connect(rstream::io::make_address(engine.address()), [&](const boost::system::error_code& error_code) { + check(!error_code, "client failed to connect with negotiated liveness"); + connected = true; + }); + io_context.run(); + test_watchdog.complete(); + engine.join(); + check(!test_watchdog.timed_out(), "missing liveness acknowledgement check timed out"); + check(connected, "client did not connect before liveness expiry"); + check(disconnected, "client remained connected without heartbeat acknowledgements"); +} + +static void check_client_tolerates_intermittent_heartbeat_loss() +{ + fake_engine engine; + engine.start([](tcp::socket& socket) { + const auto open_request = read_message(socket); + check(open_request.has_open_control_channel_req(), "missing control channel request"); + check(open_request.open_control_channel_req().has_liveness(), "control channel request did not advertise liveness"); + write_message(socket, open_control_liveness_response(1000, 2500)); + for (std::uint64_t sequence = 1; sequence <= 4; ++sequence) { + const auto heartbeat = read_message(socket); + check(heartbeat.has_heartbeat(), "client did not send a negotiated heartbeat"); + check(heartbeat.heartbeat().sequence() == sequence, "negotiated heartbeat sequence is invalid"); + if (sequence % 2 == 0) { + protobuf::Message acknowledgement; + acknowledgement.mutable_heartbeat()->set_acknowledgement(sequence); + write_message(socket, acknowledgement); + } + } + write_message(socket, close_control_response()); + wait_for_peer_close(socket); + }); + boost::asio::io_context io_context; + rstream::io_rstrm::config_client config; + config.m_no_token = true; + config.m_hearbeat = true; + config.m_heartbeat_interval_ms = 1000; + config.m_connection_timeout_ms = kControlChannelTimeoutMs; + rstream::io_rstrm::client client(io_context.get_executor(), config); + bool connected = false; + bool disconnected = false; + watchdog test_watchdog(io_context); + rstream::io_rstrm::client::control_callbacks callbacks; + callbacks.m_on_disconnection_cb = [&](const boost::system::error_code& error_code) { + check(error_code == rstream::io_rstrm::error::code::server_error, "intermittent heartbeat loss closed the connection before recovery"); + disconnected = true; + }; + boost::system::error_code callback_error; + client.set_control_callbacks(callbacks, callback_error); + check(!callback_error, "failed to install liveness callback"); + client.async_connect(rstream::io::make_address(engine.address()), [&](const boost::system::error_code& error_code) { + check(!error_code, "client failed to connect with negotiated liveness"); + connected = true; + }); + io_context.run(); + test_watchdog.complete(); + engine.join(); + check(!test_watchdog.timed_out(), "intermittent liveness check timed out"); + check(connected, "client did not connect before intermittent heartbeat loss"); + check(disconnected, "client did not process the post-recovery close response"); +} + +static void check_client_rejects_invalid_liveness_acknowledgement() +{ + fake_engine engine; + engine.start([](tcp::socket& socket) { + const auto open_request = read_message(socket); + check(open_request.has_open_control_channel_req(), "missing control channel request"); + check(open_request.open_control_channel_req().has_liveness(), "control channel request did not advertise liveness"); + write_message(socket, open_control_liveness_response(1000, 60000)); + const auto heartbeat = read_message(socket); + check(heartbeat.has_heartbeat() && heartbeat.heartbeat().sequence() == 1, "client did not send the first negotiated heartbeat"); + protobuf::Message acknowledgement; + acknowledgement.mutable_heartbeat()->set_acknowledgement(heartbeat.heartbeat().sequence() + 1); + write_message(socket, acknowledgement); + wait_for_peer_close(socket); + }); + boost::asio::io_context io_context; + rstream::io_rstrm::config_client config; + config.m_no_token = true; + config.m_hearbeat = true; + config.m_heartbeat_interval_ms = 1000; + config.m_connection_timeout_ms = kControlChannelTimeoutMs; + rstream::io_rstrm::client client(io_context.get_executor(), config); + bool connected = false; + bool disconnected = false; + watchdog test_watchdog(io_context); + rstream::io_rstrm::client::control_callbacks callbacks; + callbacks.m_on_disconnection_cb = [&](const boost::system::error_code& error_code) { + check(error_code == rstream::io_rstrm::error::code::protocol_error, "invalid heartbeat acknowledgement returned the wrong error"); + disconnected = true; + }; + boost::system::error_code callback_error; + client.set_control_callbacks(callbacks, callback_error); + check(!callback_error, "failed to install liveness callback"); + client.async_connect(rstream::io::make_address(engine.address()), [&](const boost::system::error_code& error_code) { + check(!error_code, "client failed to connect with negotiated liveness"); + connected = true; + }); + io_context.run(); + test_watchdog.complete(); + engine.join(); + check(!test_watchdog.timed_out(), "invalid liveness acknowledgement check timed out"); + check(connected, "client did not connect before invalid acknowledgement"); + check(disconnected, "client accepted an invalid heartbeat acknowledgement"); +} + +static void check_client_rejects_replayed_liveness_acknowledgement() +{ + fake_engine engine; + engine.start([](tcp::socket& socket) { + const auto open_request = read_message(socket); + check(open_request.has_open_control_channel_req(), "missing control channel request"); + write_message(socket, open_control_liveness_response(1000, 60000)); + const auto heartbeat = read_message(socket); + check(heartbeat.has_heartbeat() && heartbeat.heartbeat().sequence() == 1, "client did not send the first negotiated heartbeat"); + protobuf::Message acknowledgement; + acknowledgement.mutable_heartbeat()->set_acknowledgement(heartbeat.heartbeat().sequence()); + write_message(socket, acknowledgement); + write_message(socket, acknowledgement); + wait_for_peer_close(socket); + }); + boost::asio::io_context io_context; + rstream::io_rstrm::config_client config; + config.m_no_token = true; + config.m_hearbeat = true; + config.m_heartbeat_interval_ms = 1000; + config.m_connection_timeout_ms = kControlChannelTimeoutMs; + rstream::io_rstrm::client client(io_context.get_executor(), config); + bool connected = false; + bool disconnected = false; + watchdog test_watchdog(io_context); + rstream::io_rstrm::client::control_callbacks callbacks; + callbacks.m_on_disconnection_cb = [&](const boost::system::error_code& error_code) { + check(error_code == rstream::io_rstrm::error::code::protocol_error, "replayed heartbeat acknowledgement returned the wrong error"); + disconnected = true; + }; + boost::system::error_code callback_error; + client.set_control_callbacks(callbacks, callback_error); + check(!callback_error, "failed to install liveness callback"); + client.async_connect(rstream::io::make_address(engine.address()), [&](const boost::system::error_code& error_code) { + check(!error_code, "client failed to connect with negotiated liveness"); + connected = true; + }); + io_context.run(); + test_watchdog.complete(); + engine.join(); + check(!test_watchdog.timed_out(), "replayed liveness acknowledgement check timed out"); + check(connected, "client did not connect before replayed acknowledgement"); + check(disconnected, "client accepted a replayed heartbeat acknowledgement"); +} + +static void check_client_rejects_invalid_liveness_configuration() +{ + for (const auto interval : {999U, 300001U}) { + fake_engine engine; + engine.start([](tcp::socket& socket) { wait_for_peer_close(socket); }); + boost::asio::io_context io_context; + rstream::io_rstrm::config_client config; + config.m_no_token = true; + config.m_hearbeat = true; + config.m_heartbeat_interval_ms = interval; + config.m_connection_timeout_ms = kControlChannelTimeoutMs; + rstream::io_rstrm::client client(io_context.get_executor(), config); + bool completed = false; + watchdog test_watchdog(io_context); + client.async_connect(rstream::io::make_address(engine.address()), [&](const boost::system::error_code& error_code) { + check(error_code == rstream::io_rstrm::error::code::invalid_configuration, "invalid heartbeat interval returned the wrong error"); + completed = true; + }); + io_context.run(); + test_watchdog.complete(); + engine.join(); + check(!test_watchdog.timed_out(), "invalid heartbeat interval check timed out"); + check(completed, "invalid heartbeat interval did not complete"); + } +} + +static void check_client_rejects_invalid_server_liveness_policy() +{ + const std::array, 3> policies = {{{2000, 60000}, {1000, 999}, {1000, 900001}}}; + for (const auto& policy : policies) { + fake_engine engine; + engine.start([policy](tcp::socket& socket) { + const auto open_request = read_message(socket); + check(open_request.has_open_control_channel_req(), "missing control channel request"); + write_message(socket, open_control_liveness_response(policy.first, policy.second)); + wait_for_peer_close(socket); + }); + boost::asio::io_context io_context; + rstream::io_rstrm::config_client config; + config.m_no_token = true; + config.m_hearbeat = true; + config.m_heartbeat_interval_ms = 1000; + config.m_connection_timeout_ms = kControlChannelTimeoutMs; + rstream::io_rstrm::client client(io_context.get_executor(), config); + bool completed = false; + watchdog test_watchdog(io_context); + client.async_connect(rstream::io::make_address(engine.address()), [&](const boost::system::error_code& error_code) { + check(error_code == rstream::io_rstrm::error::code::protocol_error, "invalid server liveness policy returned the wrong error"); + completed = true; + }); + io_context.run(); + test_watchdog.complete(); + engine.join(); + check(!test_watchdog.timed_out(), "invalid server liveness policy check timed out"); + check(completed, "invalid server liveness policy did not complete"); + } + fake_engine engine; + engine.start([](tcp::socket& socket) { + const auto open_request = read_message(socket); + check(open_request.has_open_control_channel_req(), "missing control channel request"); + check(!open_request.open_control_channel_req().has_liveness(), "disabled heartbeat advertised liveness"); + write_message(socket, open_control_liveness_response(1000, 60000)); + wait_for_peer_close(socket); + }); + boost::asio::io_context io_context; + rstream::io_rstrm::config_client config; + config.m_no_token = true; + config.m_hearbeat = false; + config.m_connection_timeout_ms = kControlChannelTimeoutMs; + rstream::io_rstrm::client client(io_context.get_executor(), config); + bool completed = false; + watchdog test_watchdog(io_context); + client.async_connect(rstream::io::make_address(engine.address()), [&](const boost::system::error_code& error_code) { + check(error_code == rstream::io_rstrm::error::code::protocol_error, "unsolicited server liveness policy returned the wrong error"); + completed = true; + }); + io_context.run(); + test_watchdog.complete(); + engine.join(); + check(!test_watchdog.timed_out(), "unsolicited server liveness policy check timed out"); + check(completed, "unsolicited server liveness policy did not complete"); +} + static void check_client_can_reconnect_from_disconnection_callback() { fake_engine first_engine; @@ -1626,6 +1956,70 @@ static void check_socket_rejects_invalid_state_operations() assert(!close_error); } +static void check_error_completions_are_thread_safe() +{ + constexpr std::size_t operation_count = 256; + boost::asio::io_context io_context; + std::vector> sockets; + std::vector> clients; + std::vector> acceptors; + std::vector> accepted_peers; + std::vector> accepted_endpoints; + sockets.reserve(operation_count); + clients.reserve(operation_count); + acceptors.reserve(operation_count); + accepted_peers.reserve(operation_count); + accepted_endpoints.reserve(operation_count); + std::atomic_size_t completions = 0; + std::atomic_size_t failures = 0; + for (std::size_t index = 0; index < operation_count; ++index) { + sockets.push_back(std::make_unique(io_context.get_executor())); + rstream::io_rstrm::endpoint missing_id; + missing_id.m_server_address = rstream::io::make_address("127.0.0.1:1"); + sockets.back()->async_connect(missing_id, [&](const boost::system::error_code& error_code) { + if (error_code != rstream::io_rstrm::error::code::invalid_endpoint) { + failures.fetch_add(1, std::memory_order_relaxed); + } + completions.fetch_add(1, std::memory_order_relaxed); + }); + + rstream::io_rstrm::config_client client_config; + client_config.m_no_token = true; + clients.push_back(std::make_unique(io_context.get_executor(), client_config)); + rstream::io_rstrm::tunnel_properties properties; + clients.back()->async_create_tunnel(properties, [&](const boost::system::error_code& error_code, rstream::io_rstrm::tunnel) { + if (error_code != rstream::io_rstrm::error::code::invalid_state) { + failures.fetch_add(1, std::memory_order_relaxed); + } + completions.fetch_add(1, std::memory_order_relaxed); + }); + + rstream::io_rstrm::settings_acceptor acceptor_settings; + acceptor_settings.m_config.m_no_token = true; + acceptors.push_back(std::make_unique(io_context.get_executor(), acceptor_settings)); + accepted_peers.push_back(std::make_unique(io_context.get_executor())); + accepted_endpoints.push_back(std::make_unique()); + acceptors.back()->async_accept( + *accepted_peers.back(), + *accepted_endpoints.back(), + [&](const boost::system::error_code& error_code) { + if (error_code != rstream::io_rstrm::error::code::no_valid_endpoint) { + failures.fetch_add(1, std::memory_order_relaxed); + } + completions.fetch_add(1, std::memory_order_relaxed); + }); + } + std::array workers; + for (auto& worker : workers) { + worker = std::thread([&] { io_context.run(); }); + } + for (auto& worker : workers) { + worker.join(); + } + assert(failures.load(std::memory_order_relaxed) == 0); + assert(completions.load(std::memory_order_relaxed) == operation_count * 3); +} + static void check_async_connect_freezes_client_configuration() { boost::asio::io_context io_context; @@ -2457,12 +2851,20 @@ int main(int argc, char** argv) run_selected_check(selected, "client_rejects_malformed_tunnel_responses", check_client_rejects_malformed_tunnel_responses); run_selected_check(selected, "client_rejects_duplicate_active_tunnel_id", check_client_rejects_duplicate_active_tunnel_id); run_selected_check(selected, "client_heartbeat_and_unexpected_close_response", check_client_heartbeat_and_unexpected_close_response); + run_selected_check(selected, "client_negotiates_liveness_and_accepts_delayed_acknowledgement", check_client_negotiates_liveness_and_accepts_delayed_acknowledgement); + run_selected_check(selected, "client_expires_missing_liveness_acknowledgement", check_client_expires_missing_liveness_acknowledgement); + run_selected_check(selected, "client_tolerates_intermittent_heartbeat_loss", check_client_tolerates_intermittent_heartbeat_loss); + run_selected_check(selected, "client_rejects_invalid_liveness_acknowledgement", check_client_rejects_invalid_liveness_acknowledgement); + run_selected_check(selected, "client_rejects_replayed_liveness_acknowledgement", check_client_rejects_replayed_liveness_acknowledgement); + run_selected_check(selected, "client_rejects_invalid_liveness_configuration", check_client_rejects_invalid_liveness_configuration); + run_selected_check(selected, "client_rejects_invalid_server_liveness_policy", check_client_rejects_invalid_server_liveness_policy); run_selected_check(selected, "client_can_reconnect_from_disconnection_callback", check_client_can_reconnect_from_disconnection_callback); run_selected_check(selected, "client_drains_active_control_write_before_reconnect", check_client_drains_active_control_write_before_reconnect); run_selected_check(selected, "client_accepts_delayed_proxy_stream_and_rejects_max_streams", check_client_accepts_delayed_proxy_stream_and_rejects_max_streams); run_selected_check(selected, "client_rejects_proxy_request_for_unknown_tunnel", check_client_rejects_proxy_request_for_unknown_tunnel); run_selected_check(selected, "client_reports_redirected_proxy_failures", check_client_reports_redirected_proxy_failures); run_selected_check(selected, "socket_rejects_invalid_state_operations", check_socket_rejects_invalid_state_operations); + run_selected_check(selected, "error_completions_are_thread_safe", check_error_completions_are_thread_safe); run_selected_check(selected, "async_connect_freezes_client_configuration", check_async_connect_freezes_client_configuration); run_selected_check(selected, "async_connect_freezes_socket_configuration", check_async_connect_freezes_socket_configuration); run_selected_check(selected, "async_accept_freezes_acceptor_configuration", check_async_accept_freezes_acceptor_configuration); From 56e2a0f97ad5f696dda1ee63a1f5a7b799b757fc Mon Sep 17 00:00:00 2001 From: uartnet <140632163+uartnet@users.noreply.github.com> Date: Sat, 15 Aug 2026 01:54:34 +0200 Subject: [PATCH 3/9] fix(io): serialize remote close and TLS half-close --- bin/webtty/lib/rstream/webtty/client.cpp | 14 ++++++++---- .../io/detail/stream/stream_socket_ssl.cpp | 22 ++++++++++++++++--- 2 files changed, 29 insertions(+), 7 deletions(-) diff --git a/bin/webtty/lib/rstream/webtty/client.cpp b/bin/webtty/lib/rstream/webtty/client.cpp index 42fc08b..04bc4e6 100644 --- a/bin/webtty/lib/rstream/webtty/client.cpp +++ b/bin/webtty/lib/rstream/webtty/client.cpp @@ -1195,12 +1195,18 @@ void client::impl::on_send_message(const std::error_code& error_code, enum loop if (m_state == state::null || m_state == state::disconnected) { return; } - if (error_code) { - on_error(error_code); - } - else if (m_remote_return_code) { + if (m_remote_return_code) { finish_cmd_if_idle(); } + else if (error_code) { + if (m_state == state::connected) { + set_state(state::disconnecting); + arm_state_timer(m_settings.m_common.m_timeouts_ms.m_close); + } + if (!m_error_code) { + m_error_code = error_code; + } + } else { switch (loop) { case loop::read_std_in: diff --git a/lib/rstream/io/detail/stream/stream_socket_ssl.cpp b/lib/rstream/io/detail/stream/stream_socket_ssl.cpp index b74b1b6..b4c9e54 100644 --- a/lib/rstream/io/detail/stream/stream_socket_ssl.cpp +++ b/lib/rstream/io/detail/stream/stream_socket_ssl.cpp @@ -558,15 +558,31 @@ void stream_socket_ssl::impl:: return; } ::SSL_set_shutdown(ssl, previous_state | SSL_RECEIVED_SHUTDOWN); + auto completion_handler = rstream::core::bind_associated_handler( + std::move(handler), + [self = shared_from_this(), previous_receive, executor = m_next_layer->get_executor()](auto& associated_handler, const boost::system::error_code& error_code) mutable { +#if SSL_STREAM_USE_STRAND == 1 +#ifdef DEBUG_BUILD + assert(self->m_strand.running_in_this_thread()); +#endif +#endif + auto* native_ssl = self->m_ssl_stream.native_handle(); + const auto shutdown_state = ::SSL_get_shutdown(native_ssl); + ::SSL_set_shutdown(native_ssl, (shutdown_state & ~SSL_RECEIVED_SHUTDOWN) | previous_receive); + rstream::core::dispatch_completion_handler(executor, std::move(associated_handler), error_code); + }); + auto lifetime_handler = rstream::core::bind_handler_lifetime(shared_from_this(), std::move(completion_handler)); try { - m_ssl_stream.async_shutdown(bind_ssl_handler(rstream::core::bind_handler_lifetime(shared_from_this(), std::move(handler)))); +#if SSL_STREAM_USE_STRAND == 1 + m_ssl_stream.async_shutdown(boost::asio::bind_executor(m_strand, std::move(lifetime_handler))); +#else + m_ssl_stream.async_shutdown(std::move(lifetime_handler)); +#endif } catch (...) { ::SSL_set_shutdown(ssl, previous_state); throw; } - const auto shutdown_state = ::SSL_get_shutdown(ssl); - ::SSL_set_shutdown(ssl, (shutdown_state & ~SSL_RECEIVED_SHUTDOWN) | previous_receive); } void stream_socket_ssl::impl:: From 6970fd6f2a1452f3dd1624afc1d73de234cbcb04 Mon Sep 17 00:00:00 2001 From: uartnet <140632163+uartnet@users.noreply.github.com> Date: Sat, 15 Aug 2026 02:24:01 +0200 Subject: [PATCH 4/9] fix(webtty): drain in-flight messages on remote exit --- bin/webtty/lib/rstream/webtty/server.cpp | 63 ++++++++++++++- .../test_webtty_plain_client_runtime.cpp | 8 ++ .../test_webtty_plain_server_runtime.cpp | 76 ++++++++++++++++++- 3 files changed, 142 insertions(+), 5 deletions(-) diff --git a/bin/webtty/lib/rstream/webtty/server.cpp b/bin/webtty/lib/rstream/webtty/server.cpp index 69e475d..8edab97 100644 --- a/bin/webtty/lib/rstream/webtty/server.cpp +++ b/bin/webtty/lib/rstream/webtty/server.cpp @@ -602,6 +602,10 @@ class RSTREAM_GNUC_INTERNAL server::impl::session : public std::enable_shared_fr void on_send_message(const std::error_code& error_code, enum loop loop); + void do_shutdown_plain_send(); + + void on_shutdown_plain_send(const std::error_code& error_code); + void process_incoming_messages_loop(); void do_read_incoming_message(); @@ -678,6 +682,8 @@ class RSTREAM_GNUC_INTERNAL server::impl::session : public std::enable_shared_fr std::error_code m_error_code; + bool m_incoming_read_pending = false; + state_session_changed_signal_type m_state_changed_signal; std::set m_active_streams; @@ -1863,6 +1869,7 @@ void server::impl::session::do_send_error(const std::error_code& error_code) } log_session_rejected(error_code); set_state(state::disconnecting); + arm_state_timer(m_settings.m_common.m_timeouts_ms.m_close); rstream::webtty::protobuf::Message message; error::code code; if (error_code.category() == std::error_code(error::code{}).category()) { @@ -1896,6 +1903,7 @@ void server::impl::session::do_send_close(int code) } if (m_active_streams.empty()) { set_state(state::disconnecting); + arm_state_timer(m_settings.m_common.m_timeouts_ms.m_close); rstream::webtty::protobuf::Message message; message.mutable_close()->set_return_code(code); do_send_message(message, loop::exit); @@ -1957,7 +1965,7 @@ void server::impl::session::on_send_message(const std::error_code& error_code, e do_close_websocket(); } else { - on_close(error_code); + do_shutdown_plain_send(); } break; case loop::null: { @@ -1971,6 +1979,38 @@ void server::impl::session::on_send_message(const std::error_code& error_code, e } } +void server::impl::session::do_shutdown_plain_send() +{ +#ifdef DEBUG_BUILD + assert(m_strand.running_in_this_thread()); +#endif + if (m_state != state::disconnecting || m_websocket) { + return; + } +#ifdef RSTREAM_WITH_IO_STREAMS + auto completion_handler = std::bind(&session::on_shutdown_plain_send, shared_from_this(), std::placeholders::_1); + m_socket.async_shutdown_send(boost::asio::bind_executor(m_strand, completion_handler)); +#else + boost::system::error_code error_code; + m_socket.shutdown(boost::asio::socket_base::shutdown_send, error_code); + on_shutdown_plain_send(error_code); +#endif + do_read_incoming_message(); +} + +void server::impl::session::on_shutdown_plain_send(const std::error_code& error_code) +{ +#ifdef DEBUG_BUILD + assert(m_strand.running_in_this_thread()); +#endif + if (m_state != state::disconnecting) { + return; + } + if (error_code && !core::helpers::is_eof_error(error_code)) { + on_error(error_code); + } +} + void server::impl::session::process_incoming_messages_loop() { #ifdef DEBUG_BUILD @@ -1984,9 +2024,10 @@ void server::impl::session::do_read_incoming_message() #ifdef DEBUG_BUILD assert(m_strand.running_in_this_thread()); #endif - if (m_state == state::null || m_state == state::disconnected) { + if (m_state == state::null || m_state == state::disconnected || m_incoming_read_pending) { return; } + m_incoming_read_pending = true; m_buffer_socket.reset_size(); auto self = shared_from_this(); auto completion_handler = std::bind(&session::on_read_incoming_data, self, std::placeholders::_1); @@ -2008,11 +2049,18 @@ void server::impl::session::on_read_incoming_data(const std::error_code& error_c #ifdef DEBUG_BUILD assert(m_strand.running_in_this_thread()); #endif + assert(m_incoming_read_pending); + m_incoming_read_pending = false; if (m_state == state::null || m_state == state::disconnected) { return; } if (error_code) { - on_error(error_code); + if (m_state == state::disconnecting && core::helpers::is_eof_error(error_code)) { + on_close(std::error_code()); + } + else { + on_error(error_code); + } } else { rstream::webtty::protobuf::Message message; @@ -2182,6 +2230,10 @@ void server::impl::session::on_read_incoming_message(const rstream::webtty::prot error_code = error::code::unexpected_message; } else { + if (m_state == state::disconnecting) { + do_read_incoming_message(); + return; + } if (message_type == payload_type::kOpen) { protocol::config protocol_config; detail::convert(protocol_config, message.open().config()); @@ -2303,7 +2355,10 @@ void server::impl::session::on_process_data(const std::error_code& error_code) if (m_state == state::null || m_state == state::disconnected) { return; } - if (error_code) { + if (m_state == state::disconnecting) { + do_read_incoming_message(); + } + else if (error_code) { on_error(error_code); } else { diff --git a/test/webtty/test_webtty_plain_client_runtime.cpp b/test/webtty/test_webtty_plain_client_runtime.cpp index 608c47b..c2ad823 100644 --- a/test/webtty/test_webtty_plain_client_runtime.cpp +++ b/test/webtty/test_webtty_plain_client_runtime.cpp @@ -1,5 +1,6 @@ // See LICENSE file in the project root for license information. +#include #include #include #include @@ -516,6 +517,13 @@ class fake_early_exit_plain_server { protobuf::Message close; close.mutable_close()->set_return_code(0); write_message(socket, close); + boost::system::error_code error_code; + socket.shutdown(tcp::socket::shutdown_send, error_code); + assert(!error_code); + std::array drain_buffer{}; + while (socket.read_some(boost::asio::buffer(drain_buffer), error_code) != 0) { + } + assert(error_code == boost::asio::error::eof); } catch (...) { m_exception = std::current_exception(); diff --git a/test/webtty/test_webtty_plain_server_runtime.cpp b/test/webtty/test_webtty_plain_server_runtime.cpp index 7ce52d5..3e51e8b 100644 --- a/test/webtty/test_webtty_plain_server_runtime.cpp +++ b/test/webtty/test_webtty_plain_server_runtime.cpp @@ -3,6 +3,7 @@ #include #include #include +#include #include #include #include @@ -63,6 +64,16 @@ static void write_message(tcp::socket& socket, const protobuf::Message& message) } } +static void append_message(std::vector& dst, const protobuf::Message& message) +{ + const auto size = static_cast(message.ByteSizeLong()); + std::uint32_t frame_size = htonl(size); + const auto offset = dst.size(); + dst.resize(offset + sizeof(frame_size) + size); + std::memcpy(dst.data() + offset, &frame_size, sizeof(frame_size)); + assert(message.SerializeToArray(dst.data() + offset + sizeof(frame_size), static_cast(size))); +} + static void read_exact(tcp::socket& socket, void* data, std::size_t size) { std::size_t offset = 0; @@ -656,6 +667,46 @@ static void check_plain_server_reports_child_exit_without_stdin_eos(unsigned sho } } +static void check_plain_server_delivers_child_exit_while_messages_are_in_flight(unsigned short port) +{ + boost::asio::io_context io_context; + auto socket = connect_with_retry(io_context, port); + write_message(socket, open_message({"/bin/sh", "-c", "IFS= read -r line; exit 23"}, true)); + + auto ack = read_message(socket); + assert(ack.payload_case() == protobuf::Message::PayloadCase::kAck); + protobuf::Message heartbeat; + heartbeat.mutable_heartbeat(); + std::vector in_flight; + in_flight.reserve(512 * 1024); + for (int i = 0; i < 1000; ++i) { + append_message(in_flight, heartbeat); + } + append_message(in_flight, data_message(protobuf::Data::TYPE_STDIN, "exit\n")); + for (int i = 0; i < 50000; ++i) { + append_message(in_flight, heartbeat); + } + boost::asio::write(socket, boost::asio::buffer(in_flight)); + + bool saw_close = false; + while (!saw_close) { + auto message = read_message(socket); + switch (message.payload_case()) { + case protobuf::Message::PayloadCase::kData: + case protobuf::Message::PayloadCase::kHeartbeat: + break; + case protobuf::Message::PayloadCase::kClose: + assert(message.close().return_code() == 23); + saw_close = true; + break; + default: + std::cerr << "unexpected webtty message type: " << message.payload_case() << std::endl; + assert(false); + break; + } + } +} + static void check_plain_server_e2e_forwards_stdin_to_child_process() { std::error_code error_code; @@ -972,9 +1023,31 @@ static void check_plain_server_cancel_keeps_active_child_resources_alive() assert(out.payload_case() == protobuf::Message::PayloadCase::kData); assert(out.data().type() == protobuf::Data::TYPE_STDOUT); assert(out.data().data() == "active"); - server.stop(); + std::exception_ptr stop_exception; + std::thread stop_thread([&] { + try { + server.stop(); + } + catch (...) { + stop_exception = std::current_exception(); + } + }); + bool saw_close = false; + while (!saw_close) { + auto message = read_message(socket); + if (message.payload_case() == protobuf::Message::PayloadCase::kClose) { + saw_close = true; + } + else { + assert(message.payload_case() == protobuf::Message::PayloadCase::kData || message.payload_case() == protobuf::Message::PayloadCase::kHeartbeat); + } + } boost::system::error_code ignored; socket.close(ignored); + stop_thread.join(); + if (stop_exception) { + std::rethrow_exception(stop_exception); + } } } @@ -1002,6 +1075,7 @@ int main(int argc, char** argv) run_check("tty environment and workdir", [&server] { check_plain_server_applies_tty_environment_and_workdir(server.port()); }); run_check("stdin forwarding", [&server] { check_plain_server_forwards_stdin_to_child_process(server.port()); }); run_check("child exit before stdin EOS", [&server] { check_plain_server_reports_child_exit_without_stdin_eos(server.port()); }); + run_check("child exit while messages are in flight", [&server] { check_plain_server_delivers_child_exit_while_messages_are_in_flight(server.port()); }); run_check("E2E stdin forwarding", check_plain_server_e2e_forwards_stdin_to_child_process); run_check("client credential verification", check_plain_server_e2e_accepts_client_credential_verifier); run_check("missing session key rejection", check_plain_server_e2e_rejects_missing_session_key_grant); From cbd73aa0c5738f759ea58552aefba9513d4c101f Mon Sep 17 00:00:00 2001 From: uartnet <140632163+uartnet@users.noreply.github.com> Date: Sat, 15 Aug 2026 05:42:35 +0200 Subject: [PATCH 5/9] fix(io): prevent child stdin SIGPIPE --- bin/ncat/lib/rstream/ncat/server.cpp | 37 +++++++ .../lib/rstream/webtty/detail/process.hpp | 29 ++++- bin/webtty/lib/rstream/webtty/stream.cpp | 51 +++++++-- bin/webtty/lib/rstream/webtty/stream.hpp | 22 +++- lib/rstream/core/posix/child_stdin.cpp | 88 +++++++++++++++ lib/rstream/core/posix/child_stdin.hpp | 48 +++++++++ test/ncat/test_ncat_runtime.cpp | 49 ++++++++- test/webtty/test_webtty_terminal_stream.cpp | 101 +++++++++++++++++- 8 files changed, 408 insertions(+), 17 deletions(-) create mode 100644 lib/rstream/core/posix/child_stdin.cpp create mode 100644 lib/rstream/core/posix/child_stdin.hpp diff --git a/bin/ncat/lib/rstream/ncat/server.cpp b/bin/ncat/lib/rstream/ncat/server.cpp index 067d6c3..22b6e86 100644 --- a/bin/ncat/lib/rstream/ncat/server.cpp +++ b/bin/ncat/lib/rstream/ncat/server.cpp @@ -44,6 +44,13 @@ #include #include #endif +#ifndef _WIN32 +#if __has_include() +#include +#else +#include +#endif +#endif #include #include @@ -52,6 +59,8 @@ #ifdef _WIN32 #include #include +#else +#include #endif // clang-format on @@ -64,6 +73,9 @@ #include #include #include +#ifndef _WIN32 +#include +#endif #include #ifdef RSTREAM_WITH_IO_STREAMS #include @@ -460,7 +472,11 @@ class RSTREAM_GNUC_INTERNAL server::impl::session_exec : public session, public async_run_completion_handler m_handler; +#ifdef _WIN32 boost::process::async_pipe m_child_stdin; +#else + rstream::core::posix::child_stdin m_child_stdin; +#endif boost::process::async_pipe m_child_stdout; @@ -1179,7 +1195,11 @@ server::impl::session_exec::session_exec(socket_type&& downstream_socket, const m_exec(exec), m_logger({"rstream", "ncat", "session", fmt::format("#{}", session_id)}), m_state(state::null), +#ifdef _WIN32 m_child_stdin(get_io_context(m_executor)), +#else + m_child_stdin(m_executor), +#endif m_child_stdout(get_io_context(m_executor)), m_child_stderr(get_io_context(m_executor)), m_child_stdout_eos(false), @@ -1284,9 +1304,14 @@ void server::impl::session_exec::start_child() m_logger->trace("starting child process [shell: {} | cmd: {}]", shell, m_exec.m_cmd); m_child = std::make_shared(shell, boost::process::args(args), +#ifdef _WIN32 boost::process::std_in m_child_stdout, +#else + boost::process::posix::fd.bind(STDIN_FILENO, m_child_stdin.child_native_handle()), + boost::process::std_out > m_child_stdout, +#endif boost::process::std_err > m_child_stderr, boost::process::on_exit = completion_handler, get_io_context(m_executor)); @@ -1314,9 +1339,14 @@ void server::impl::session_exec::start_child() m_logger->trace("starting child process [exe: {} | args: {}]", exe, args_stream.str()); m_child = std::make_shared(exe, boost::process::args(args), +#ifdef _WIN32 boost::process::std_in m_child_stdout, +#else + boost::process::posix::fd.bind(STDIN_FILENO, m_child_stdin.child_native_handle()), + boost::process::std_out > m_child_stdout, +#endif boost::process::std_err > m_child_stderr, boost::process::on_exit = completion_handler, get_io_context(m_executor)); @@ -1326,6 +1356,9 @@ void server::impl::session_exec::start_child() catch (...) { exception_ptr = std::current_exception(); } +#ifndef _WIN32 + m_child_stdin.close_child_end(); +#endif if (exception_ptr) { try { std::rethrow_exception(exception_ptr); @@ -1437,7 +1470,11 @@ void server::impl::session_exec::do_write_child() auto completion_handler = [self = shared_from_this(), buffer](const boost::system::error_code& error_code, std::size_t size) { self->on_write_child(error_code, size); }; +#ifdef _WIN32 boost::asio::async_write(m_child_stdin, core::helpers::const_memory_sequence(*buffer), boost::asio::bind_executor(m_strand, std::move(completion_handler))); +#else + boost::asio::async_write(m_child_stdin.stream(), core::helpers::const_memory_sequence(*buffer), boost::asio::bind_executor(m_strand, std::move(completion_handler))); +#endif } void server::impl::session_exec::on_write_child(const boost::system::error_code& error_code, std::size_t size) diff --git a/bin/webtty/lib/rstream/webtty/detail/process.hpp b/bin/webtty/lib/rstream/webtty/detail/process.hpp index 2a83505..7876f7e 100644 --- a/bin/webtty/lib/rstream/webtty/detail/process.hpp +++ b/bin/webtty/lib/rstream/webtty/detail/process.hpp @@ -29,6 +29,13 @@ #else #include #endif +#ifndef _WIN32 +#if __has_include() +#include +#else +#include +#endif +#endif #include #include @@ -109,10 +116,24 @@ std::shared_ptr make_child(stream::ptr stream_ptr, Args&& } else { auto stream_ptr_pipe = std::dynamic_pointer_cast(stream_ptr); - child = std::make_shared(boost::process::std_out > stream_ptr_pipe->stream(stream::type::std_out), - boost::process::std_err > stream_ptr_pipe->stream(stream::type::std_err), - boost::process::std_in < stream_ptr_pipe->stream(stream::type::std_in), - std::forward(args)...); +#ifdef _WIN32 + child = std::make_shared(boost::process::std_out > stream_ptr_pipe->output_stream(stream::type::std_out), + boost::process::std_err > stream_ptr_pipe->output_stream(stream::type::std_err), + boost::process::std_in < stream_ptr_pipe->input_stream(), + std::forward(args)...); +#else + try { + child = std::make_shared(boost::process::std_out > stream_ptr_pipe->output_stream(stream::type::std_out), + boost::process::std_err > stream_ptr_pipe->output_stream(stream::type::std_err), + boost::process::posix::fd.bind(STDIN_FILENO, stream_ptr_pipe->child_stdin_native_handle()), + std::forward(args)...); + } + catch (...) { + stream_ptr_pipe->child_spawned(); + throw; + } + stream_ptr_pipe->child_spawned(); +#endif } return child; } diff --git a/bin/webtty/lib/rstream/webtty/stream.cpp b/bin/webtty/lib/rstream/webtty/stream.cpp index 5a59e25..988f12c 100644 --- a/bin/webtty/lib/rstream/webtty/stream.cpp +++ b/bin/webtty/lib/rstream/webtty/stream.cpp @@ -6,6 +6,7 @@ #include #include #include +#include #include @@ -40,7 +41,11 @@ ptr make_stream(const executor_type& executor, backend backend) pipe::pipe(const executor_type& executor) : base(backend::pipe), +#ifdef _WIN32 m_std_in(executor.context()), +#else + m_std_in(executor), +#endif m_std_out(executor.context()), m_std_err(executor.context()) { @@ -53,12 +58,22 @@ pipe::~pipe() void pipe::async_read_some(const boost::asio::mutable_buffer& buffer, type type, async_read_some_completion_handler&& handler) { - stream(type).async_read_some(buffer, std::move(handler)); + if (type == type::std_in) { + input_stream().async_read_some(buffer, std::move(handler)); + } + else { + output_stream(type).async_read_some(buffer, std::move(handler)); + } } void pipe::async_write(const boost::asio::const_buffer& buffer, type type, async_write_completion_handler&& handler) { - boost::asio::async_write(stream(type), buffer, std::move(handler)); + if (type == type::std_in) { + boost::asio::async_write(input_stream(), buffer, std::move(handler)); + } + else { + boost::asio::async_write(output_stream(type), buffer, std::move(handler)); + } } void pipe::close() @@ -71,22 +86,40 @@ void pipe::close() void pipe::close(type type) { boost::system::error_code tmp; - stream(type).close(tmp); + if (type == type::std_in) { + m_std_in.close(tmp); + } + else { + output_stream(type).close(tmp); + } } -pipe::stream_type& pipe::stream(type type) +pipe::stream_type& pipe::output_stream(type type) { - if (type == type::std_in) { - return m_std_in; - } - else if (type == type::std_out) { + if (type == type::std_out) { return m_std_out; } - else { + else if (type == type::std_err) { return m_std_err; } + throw std::invalid_argument("stdin is not an output stream"); } +pipe::input_stream_type& pipe::input_stream() +{ +#ifdef _WIN32 + return m_std_in; +#else + return m_std_in.stream(); +#endif +} + +#ifndef _WIN32 +int pipe::child_stdin_native_handle() { return m_std_in.child_native_handle(); } + +void pipe::child_spawned() { m_std_in.close_child_end(); } +#endif + #ifdef _WIN32 namespace { diff --git a/bin/webtty/lib/rstream/webtty/stream.hpp b/bin/webtty/lib/rstream/webtty/stream.hpp index 18e4823..25591ba 100644 --- a/bin/webtty/lib/rstream/webtty/stream.hpp +++ b/bin/webtty/lib/rstream/webtty/stream.hpp @@ -35,6 +35,9 @@ #endif #include +#ifndef _WIN32 +#include +#endif #include "error.hpp" #include "webtty.hpp" @@ -97,12 +100,25 @@ class pty { class pipe : public base { public: using stream_type = boost::process::async_pipe; +#ifdef _WIN32 + using input_stream_type = stream_type; +#else + using input_stream_type = rstream::core::posix::child_stdin::stream_type; +#endif pipe(const executor_type& executor); virtual ~pipe(); - stream_type& stream(type type); + stream_type& output_stream(type type); + + input_stream_type& input_stream(); + +#ifndef _WIN32 + int child_stdin_native_handle(); + + void child_spawned(); +#endif void async_read_some(const boost::asio::mutable_buffer& buffer, type type, async_read_some_completion_handler&& handler) override; @@ -113,7 +129,11 @@ class pipe : public base { void close(type type); private: +#ifdef _WIN32 stream_type m_std_in; +#else + rstream::core::posix::child_stdin m_std_in; +#endif stream_type m_std_out; diff --git a/lib/rstream/core/posix/child_stdin.cpp b/lib/rstream/core/posix/child_stdin.cpp new file mode 100644 index 0000000..c657a7e --- /dev/null +++ b/lib/rstream/core/posix/child_stdin.cpp @@ -0,0 +1,88 @@ +// See LICENSE file in the project root for license information. + +#include "child_stdin.hpp" + +#ifndef _WIN32 + +#include +#include + +#include +#include + +#include +#include +#include + +namespace rstream { +namespace core { +namespace posix { + +namespace { + +void set_close_on_exec(int descriptor) +{ + const auto flags = ::fcntl(descriptor, F_GETFD); + if (flags == -1 || ::fcntl(descriptor, F_SETFD, flags | FD_CLOEXEC) == -1) { + throw boost::system::system_error(errno, boost::system::system_category(), "fcntl(FD_CLOEXEC)"); + } +} + +void disable_sigpipe(int descriptor) +{ +#ifdef SO_NOSIGPIPE + constexpr int enabled = 1; + if (::setsockopt(descriptor, SOL_SOCKET, SO_NOSIGPIPE, &enabled, sizeof(enabled)) == -1) { + throw boost::system::system_error(errno, boost::system::system_category(), "setsockopt(SO_NOSIGPIPE)"); + } +#else + (void)descriptor; +#endif +} + +} // namespace + +child_stdin::child_stdin(const boost::asio::any_io_executor& executor) + : m_parent(executor), + m_child(executor) +{ + boost::asio::local::connect_pair(m_parent, m_child); + if (m_child.native_handle() == STDIN_FILENO) { + std::swap(m_parent, m_child); + } + set_close_on_exec(m_parent.native_handle()); + set_close_on_exec(m_child.native_handle()); + disable_sigpipe(m_parent.native_handle()); +} + +child_stdin::~child_stdin() +{ + boost::system::error_code ignored; + close(ignored); +} + +child_stdin::stream_type& child_stdin::stream() { return m_parent; } + +int child_stdin::child_native_handle() { return m_child.native_handle(); } + +void child_stdin::close_child_end() +{ + boost::system::error_code ignored; + m_child.close(ignored); +} + +void child_stdin::close(boost::system::error_code& error_code) +{ + error_code.clear(); + boost::system::error_code parent_error; + boost::system::error_code child_error; + m_parent.close(parent_error); + m_child.close(child_error); + error_code = parent_error ? parent_error : child_error; +} + +} // namespace posix +} // namespace core +} // namespace rstream + +#endif diff --git a/lib/rstream/core/posix/child_stdin.hpp b/lib/rstream/core/posix/child_stdin.hpp new file mode 100644 index 0000000..2c931de --- /dev/null +++ b/lib/rstream/core/posix/child_stdin.hpp @@ -0,0 +1,48 @@ +// See LICENSE file in the project root for license information. + +#pragma once + +#ifndef _WIN32 + +#include +#include + +namespace rstream { +namespace core { +namespace posix { + +// A socket pair keeps child standard-input writes on Asio's per-operation +// SIGPIPE-safe socket path without changing the process-wide signal policy. +class child_stdin { + public: + using stream_type = boost::asio::local::stream_protocol::socket; + + explicit child_stdin(const boost::asio::any_io_executor& executor); + + ~child_stdin(); + + child_stdin(const child_stdin&) = delete; + child_stdin& operator=(const child_stdin&) = delete; + + child_stdin(child_stdin&&) = delete; + child_stdin& operator=(child_stdin&&) = delete; + + stream_type& stream(); + + int child_native_handle(); + + void close_child_end(); + + void close(boost::system::error_code& error_code); + + private: + stream_type m_parent; + + stream_type m_child; +}; + +} // namespace posix +} // namespace core +} // namespace rstream + +#endif diff --git a/test/ncat/test_ncat_runtime.cpp b/test/ncat/test_ncat_runtime.cpp index 0ed1b86..c7fd229 100644 --- a/test/ncat/test_ncat_runtime.cpp +++ b/test/ncat/test_ncat_runtime.cpp @@ -2,7 +2,9 @@ #include #include +#include #include +#include #include #include #include @@ -19,6 +21,7 @@ #include #include +#include #include #include @@ -510,6 +513,46 @@ static void check_exec_server_pipes_downstream_to_child() assert(response.find("exec") != std::string::npos); } +static int run_exec_server_write_after_child_closed_stdin() +{ + std::signal(SIGPIPE, SIG_DFL); + const auto port = unused_tcp_port(); + rstream::ncat::server::config config = { + .m_local = rstream::io::address(std::string("127.0.0.1:") + std::to_string(port)), + .m_remote = rstream::ncat::server::exec{.m_shell = true, .m_cmd = "exec 0<&-; printf ready; sleep 30"}, + }; + ncat_server_fixture server(config); + server.start(); + boost::asio::io_context io_context; + auto socket = connect_with_retry(io_context, server.port()); + std::array ready{}; + boost::asio::read(socket, boost::asio::buffer(ready)); + assert(std::string(ready.data(), ready.size()) == "ready"); + boost::system::error_code error_code; + boost::asio::write(socket, boost::asio::buffer(std::string("payload")), error_code); + std::array response{}; + socket.read_some(boost::asio::buffer(response), error_code); + assert(error_code); + server.stop(); + return 0; +} + +static void check_exec_server_write_after_child_closed_stdin_does_not_raise_sigpipe(const char* executable) +{ + const auto pid = ::fork(); + require_posix(pid != -1, "fork failed"); + if (pid == 0) { + ::execl(executable, executable, "--exec-write-after-child-closed-stdin", static_cast(nullptr)); + ::_exit(127); + } + int status = 0; + while (::waitpid(pid, &status, 0) == -1) { + require_posix(errno == EINTR, "waitpid failed"); + } + assert(WIFEXITED(status)); + assert(WEXITSTATUS(status) == 0); +} + static void check_proxy_server_pipes_downstream_to_upstream() { echo_upstream upstream; @@ -685,8 +728,10 @@ static void check_exec_server_cancel_keeps_active_child_alive_until_completion() int main(int argc, char** argv) { - (void)argc; - (void)argv; + if (argc == 2 && std::string(argv[1]) == "--exec-write-after-child-closed-stdin") { + return run_exec_server_write_after_child_closed_stdin(); + } + check_exec_server_write_after_child_closed_stdin_does_not_raise_sigpipe(argv[0]); check_error_category(); check_exec_server_pipes_downstream_to_child(); check_proxy_server_pipes_downstream_to_upstream(); diff --git a/test/webtty/test_webtty_terminal_stream.cpp b/test/webtty/test_webtty_terminal_stream.cpp index 802a12f..e0e760d 100644 --- a/test/webtty/test_webtty_terminal_stream.cpp +++ b/test/webtty/test_webtty_terminal_stream.cpp @@ -4,6 +4,7 @@ #include #include +#include #include #include #include @@ -28,6 +29,7 @@ #ifdef _WIN32 #include #else +#include #include #ifdef __APPLE__ #include @@ -240,15 +242,112 @@ static void check_pipe_stream_lifecycle() auto stream_ptr = stream::make_stream(io_context.get_executor(), stream::backend::pipe); assert(stream_ptr); assert(stream_ptr->backend() == stream::backend::pipe); - assert(std::dynamic_pointer_cast(stream_ptr)); + auto pipe = std::dynamic_pointer_cast(stream_ptr); + assert(pipe); + bool invalid_output_rejected = false; + try { + (void)pipe->output_stream(stream::type::std_in); + } + catch (const std::invalid_argument&) { + invalid_output_rejected = true; + } + assert(invalid_output_rejected); stream_ptr->close(); stream_ptr->close(); } +#ifndef _WIN32 +static int run_pipe_write_after_child_exit() +{ + std::signal(SIGPIPE, SIG_DFL); + boost::asio::io_context io_context; + auto stream_ptr = stream::make_stream(io_context.get_executor(), stream::backend::pipe); + auto child = rstream::webtty::detail::process::make_child( + stream_ptr, + boost::process::exe("/bin/sh"), + boost::process::args(std::vector{"-c", "exit 0"})); + child->wait(); + std::vector payload(1024 * 1024, 'x'); + std::error_code write_error; + std::size_t completions = 0; + stream_ptr->async_write(boost::asio::buffer(payload), stream::type::std_in, [&](const std::error_code& error_code, std::size_t) { + write_error = error_code; + ++completions; + }); + io_context.run(); + assert(completions == 1); + assert(write_error); + return 0; +} + +static void check_pipe_subprocess(const char* executable, const char* mode) +{ + boost::process::child child( + boost::process::exe(executable), + boost::process::args(std::vector{mode})); + child.wait(); + assert(child.exit_code() == 0); +} + +static void check_pipe_write_after_child_exit_does_not_raise_sigpipe(const char* executable) +{ + check_pipe_subprocess(executable, "--pipe-write-after-child-exit"); + check_pipe_subprocess(executable, "--pipe-write-after-child-exit-with-closed-stdin"); +} + +static std::size_t open_descriptor_count() +{ + auto directory = ::opendir("/dev/fd"); + assert(directory != nullptr); + std::size_t count = 0; + while (const auto* entry = ::readdir(directory)) { + if (entry->d_name[0] != '.') { + ++count; + } + } + assert(::closedir(directory) == 0); + return count; +} + +static void spawn_pipe_child_once() +{ + boost::asio::io_context io_context; + auto stream_ptr = stream::make_stream(io_context.get_executor(), stream::backend::pipe); + auto child = rstream::webtty::detail::process::make_child( + stream_ptr, + boost::process::exe("/bin/sh"), + boost::process::args(std::vector{"-c", "exit 0"})); + child->wait(); + stream_ptr->close(); +} + +static void check_pipe_child_spawn_does_not_leak_descriptors() +{ + spawn_pipe_child_once(); + const auto before = open_descriptor_count(); + for (std::size_t iteration = 0; iteration < 64; ++iteration) { + spawn_pipe_child_once(); + } + assert(open_descriptor_count() == before); +} +#endif + int main(int argc, char** argv) { +#ifndef _WIN32 + if (argc == 2 && std::string(argv[1]) == "--pipe-write-after-child-exit") { + return run_pipe_write_after_child_exit(); + } + if (argc == 2 && std::string(argv[1]) == "--pipe-write-after-child-exit-with-closed-stdin") { + assert(::close(STDIN_FILENO) == 0); + return run_pipe_write_after_child_exit(); + } + check_pipe_write_after_child_exit_does_not_raise_sigpipe(argv[0]); + check_pipe_child_spawn_does_not_leak_descriptors(); +#else (void)argc; (void)argv; +#endif check_pipe_stream_lifecycle(); #ifdef _WIN32 check_windows_pty_rejects_overlapping_writes(); From 25981b5f12063c1f67a043594b397bb171d7521f Mon Sep 17 00:00:00 2001 From: uartnet <140632163+uartnet@users.noreply.github.com> Date: Sat, 15 Aug 2026 05:42:42 +0200 Subject: [PATCH 6/9] test(runtime): preserve streams across liveness expiry --- .../test_io_rstrm_control_channel.cpp | 114 ++++++++++++++++++ 1 file changed, 114 insertions(+) diff --git a/test/io/io-rstrm/test_io_rstrm_control_channel.cpp b/test/io/io-rstrm/test_io_rstrm_control_channel.cpp index 7779c72..0a0913f 100644 --- a/test/io/io-rstrm/test_io_rstrm_control_channel.cpp +++ b/test/io/io-rstrm/test_io_rstrm_control_channel.cpp @@ -15,6 +15,7 @@ #include #include #include +#include #include #include @@ -223,6 +224,10 @@ static protobuf::Message close_control_response() return response; } +static protobuf::Message proxy_connection_request(const std::string& stream_id, const std::string& tunnel_id); + +static void write_proxy_response_if_needed(tcp::socket& socket, const protobuf::Message& request); + static protobuf::Message close_tunnel_response(const std::string& tunnel_id) { protobuf::Message response; @@ -1292,6 +1297,114 @@ static void check_client_tolerates_intermittent_heartbeat_loss() check(disconnected, "client did not process the post-recovery close response"); } +static void check_established_stream_survives_control_liveness_timeout() +{ + constexpr std::string_view request_payload = "after-control-loss"; + constexpr std::string_view reply_payload = "stream-still-alive"; + fake_engine engine; + engine.start_multi([request_payload, reply_payload](tcp::acceptor& acceptor) { + auto control = accept_connection(acceptor); + auto open_request = read_message(control); + check(open_request.has_open_control_channel_req(), "missing control channel request"); + check(open_request.open_control_channel_req().has_liveness(), "control channel request did not advertise liveness"); + write_message(control, open_control_liveness_response(1000, 2500)); + bool tunnel_opened = false; + while (!tunnel_opened) { + const auto message = read_message(control); + if (message.has_heartbeat()) { + protobuf::Message acknowledgement; + acknowledgement.mutable_heartbeat()->set_acknowledgement(message.heartbeat().sequence()); + write_message(control, acknowledgement); + } + else { + check(message.has_open_tunnel_req(), "unexpected message before tunnel creation"); + write_message(control, open_tunnel_response(message.open_tunnel_req())); + tunnel_opened = true; + } + } + write_message(control, proxy_connection_request("stream-after-control-loss", "tunnel-1")); + auto stream = accept_connection(acceptor); + auto handshake = read_message(stream); + check(handshake.has_proxy_req(), "missing proxy stream handshake"); + check(handshake.proxy_req().stream_id() == "stream-after-control-loss", "proxy stream handshake used the wrong stream ID"); + write_proxy_response_if_needed(stream, handshake); + bool stream_acknowledged = false; + while (!stream_acknowledged) { + const auto message = read_message(control); + if (message.has_heartbeat()) { + protobuf::Message acknowledgement; + acknowledgement.mutable_heartbeat()->set_acknowledgement(message.heartbeat().sequence()); + write_message(control, acknowledgement); + } + else { + check(message.has_proxy_conn_rsp(), "unexpected message before proxy stream acknowledgement"); + check(!message.proxy_conn_rsp().has_error(), "proxy stream was rejected"); + stream_acknowledged = true; + } + } + wait_for_peer_close(control); + boost::asio::write(stream, boost::asio::buffer(std::string(request_payload))); + std::vector reply(reply_payload.size()); + boost::asio::read(stream, boost::asio::buffer(reply)); + check(std::string(reply.data(), reply.size()) == reply_payload, "established stream stopped after control liveness timeout"); + }); + boost::asio::io_context io_context; + rstream::io_rstrm::config_client config; + config.m_no_token = true; + config.m_hearbeat = true; + config.m_heartbeat_interval_ms = 1000; + config.m_connection_timeout_ms = kControlChannelTimeoutMs; + rstream::io_rstrm::client client(io_context.get_executor(), config); + auto tunnel = std::make_shared(); + auto accepted = std::make_shared(io_context.get_executor()); + auto endpoint = std::make_shared(); + auto request_buffer = std::make_shared>(request_payload.size()); + bool disconnected = false; + bool stream_replied = false; + watchdog test_watchdog(io_context); + rstream::io_rstrm::client::control_callbacks callbacks; + callbacks.m_on_disconnection_cb = [&](const boost::system::error_code& error_code) { + check(error_code == rstream::io_rstrm::error::code::operation_timeout, "control liveness timeout returned the wrong error"); + disconnected = true; + }; + boost::system::error_code callback_error; + client.set_control_callbacks(callbacks, callback_error); + check(!callback_error, "failed to install liveness callback"); + client.async_connect(rstream::io::make_address(engine.address()), [&](const boost::system::error_code& error_code) { + check(!error_code, "client failed to connect with negotiated liveness"); + rstream::io_rstrm::tunnel_properties properties; + properties.m_name = "api"; + properties.m_type = "bytestream"; + client.async_create_tunnel(properties, [&, tunnel, accepted, endpoint, request_buffer](const boost::system::error_code& create_error, rstream::io_rstrm::tunnel created_tunnel) { + check(!create_error, "failed to create tunnel before liveness timeout"); + *tunnel = std::move(created_tunnel); + tunnel->async_accept(*accepted, *endpoint, [&, accepted, request_buffer](const boost::system::error_code& accept_error) { + check(!accept_error, "failed to accept stream before liveness timeout"); + boost::asio::async_read(*accepted, boost::asio::buffer(*request_buffer), [&, accepted, request_buffer](const boost::system::error_code& read_error, std::size_t read) { + check(!read_error, "established stream read failed after control liveness timeout"); + check(read == request_buffer->size(), "established stream read was truncated after control liveness timeout"); + check(std::string(request_buffer->data(), request_buffer->size()) == request_payload, "established stream payload changed after control liveness timeout"); + auto reply = std::make_shared(reply_payload); + boost::asio::async_write(*accepted, boost::asio::buffer(*reply), [&, accepted, reply](const boost::system::error_code& write_error, std::size_t written) { + check(!write_error, "established stream write failed after control liveness timeout"); + check(written == reply->size(), "established stream reply was truncated after control liveness timeout"); + stream_replied = true; + boost::system::error_code close_error; + accepted->close(close_error); + check(!close_error, "failed to close established stream after liveness test"); + }); + }); + }); + }); + }); + io_context.run(); + test_watchdog.complete(); + engine.join(); + check(!test_watchdog.timed_out(), "established stream liveness check timed out"); + check(disconnected, "control channel did not expire after missing heartbeat acknowledgements"); + check(stream_replied, "established stream did not survive control liveness timeout"); +} + static void check_client_rejects_invalid_liveness_acknowledgement() { fake_engine engine; @@ -2854,6 +2967,7 @@ int main(int argc, char** argv) run_selected_check(selected, "client_negotiates_liveness_and_accepts_delayed_acknowledgement", check_client_negotiates_liveness_and_accepts_delayed_acknowledgement); run_selected_check(selected, "client_expires_missing_liveness_acknowledgement", check_client_expires_missing_liveness_acknowledgement); run_selected_check(selected, "client_tolerates_intermittent_heartbeat_loss", check_client_tolerates_intermittent_heartbeat_loss); + run_selected_check(selected, "established_stream_survives_control_liveness_timeout", check_established_stream_survives_control_liveness_timeout); run_selected_check(selected, "client_rejects_invalid_liveness_acknowledgement", check_client_rejects_invalid_liveness_acknowledgement); run_selected_check(selected, "client_rejects_replayed_liveness_acknowledgement", check_client_rejects_replayed_liveness_acknowledgement); run_selected_check(selected, "client_rejects_invalid_liveness_configuration", check_client_rejects_invalid_liveness_configuration); From d754a645aad7cab2e252f05caafe202b05058643 Mon Sep 17 00:00:00 2001 From: uartnet <140632163+uartnet@users.noreply.github.com> Date: Sat, 15 Aug 2026 05:50:34 +0200 Subject: [PATCH 7/9] test(runtime): preserve streams while acceptors reconnect --- .../test_io_rstrm_control_channel.cpp | 123 ++++++++++++++++++ 1 file changed, 123 insertions(+) diff --git a/test/io/io-rstrm/test_io_rstrm_control_channel.cpp b/test/io/io-rstrm/test_io_rstrm_control_channel.cpp index 0a0913f..3253bbd 100644 --- a/test/io/io-rstrm/test_io_rstrm_control_channel.cpp +++ b/test/io/io-rstrm/test_io_rstrm_control_channel.cpp @@ -170,6 +170,19 @@ static void wait_for_peer_close(tcp::socket& socket) } } +static protobuf::Message read_control_message_and_acknowledge_heartbeats(tcp::socket& socket) +{ + for (;;) { + auto message = read_message(socket); + if (!message.has_heartbeat()) { + return message; + } + protobuf::Message acknowledgement; + acknowledgement.mutable_heartbeat()->set_acknowledgement(message.heartbeat().sequence()); + write_message(socket, acknowledgement); + } +} + static void wait_until_ready(const std::atomic_bool& ready, const std::string& timeout_message) { const auto deadline = std::chrono::steady_clock::now() + kFakeEngineIoTimeout; @@ -1405,6 +1418,115 @@ static void check_established_stream_survives_control_liveness_timeout() check(stream_replied, "established stream did not survive control liveness timeout"); } +static void check_acceptor_reconnects_after_liveness_timeout_without_breaking_established_stream() +{ + constexpr std::string_view request_payload = "after-acceptor-reconnect"; + constexpr std::string_view reply_payload = "accepted-stream-survived"; + std::atomic_bool second_tunnel_online = false; + fake_engine engine; + engine.start_multi([request_payload, reply_payload, &second_tunnel_online](tcp::acceptor& network_acceptor) { + auto first_control = accept_connection(network_acceptor); + auto open_request = read_message(first_control); + check(open_request.has_open_control_channel_req(), "missing first acceptor control request"); + write_message(first_control, open_control_liveness_response(1000, 2500)); + auto tunnel_request = read_control_message_and_acknowledge_heartbeats(first_control); + check(tunnel_request.has_open_tunnel_req(), "missing first acceptor tunnel request"); + write_message(first_control, open_tunnel_response(tunnel_request.open_tunnel_req())); + write_message(first_control, proxy_connection_request("accepted-before-control-loss", "tunnel-1")); + auto stream = accept_connection(network_acceptor); + auto handshake = read_message(stream); + check(handshake.has_proxy_req(), "missing accepted-stream handshake"); + check(handshake.proxy_req().stream_id() == "accepted-before-control-loss", "accepted-stream handshake used the wrong stream ID"); + write_proxy_response_if_needed(stream, handshake); + auto stream_acknowledgement = read_control_message_and_acknowledge_heartbeats(first_control); + check(stream_acknowledgement.has_proxy_conn_rsp(), "missing accepted-stream acknowledgement"); + check(!stream_acknowledgement.proxy_conn_rsp().has_error(), "acceptor rejected the established stream"); + wait_for_peer_close(first_control); + auto second_control = accept_connection(network_acceptor); + open_request = read_message(second_control); + check(open_request.has_open_control_channel_req(), "acceptor did not reconnect after liveness timeout"); + write_message(second_control, open_control_liveness_response(1000, 2500)); + tunnel_request = read_control_message_and_acknowledge_heartbeats(second_control); + check(tunnel_request.has_open_tunnel_req(), "acceptor did not recreate its tunnel after reconnecting"); + write_message(second_control, open_tunnel_response(tunnel_request.open_tunnel_req())); + wait_until_ready(second_tunnel_online, "acceptor did not report its recreated tunnel online"); + boost::asio::write(stream, boost::asio::buffer(std::string(request_payload))); + std::vector reply(reply_payload.size()); + boost::asio::read(stream, boost::asio::buffer(reply)); + check(std::string(reply.data(), reply.size()) == reply_payload, "accepted stream stopped after acceptor reconnection"); + auto close_request = read_control_message_and_acknowledge_heartbeats(second_control); + check(close_request.has_close_control_channel_req(), "acceptor did not close the reconnected control channel cleanly"); + write_message(second_control, close_control_response()); + }); + boost::asio::io_context io_context; + rstream::io_rstrm::settings_acceptor settings; + settings.m_config.m_no_token = true; + settings.m_config.m_hearbeat = true; + settings.m_config.m_heartbeat_interval_ms = 1000; + settings.m_config.m_connection_timeout_ms = kControlChannelTimeoutMs; + settings.m_auto_reconnect = true; + settings.m_reconnect_timeout_ms = 1; + settings.m_auto_recreate_tunnel = true; + settings.m_recreate_tunnel_timeout_ms = 1; + settings.m_tunnel_properties.m_type = "bytestream"; + settings.m_tunnel_properties.m_publish = true; + rstream::io_rstrm::acceptor network_acceptor(io_context.get_executor(), settings); + rstream::io_rstrm::endpoint endpoint; + endpoint.m_id_name = "api"; + endpoint.m_server_address = rstream::io::make_address(engine.address()); + boost::system::error_code bind_error; + network_acceptor.bind(endpoint, bind_error); + check(!bind_error, "failed to bind reconnecting acceptor"); + std::size_t online_status_count = 0; + std::size_t disconnected_count = 0; + rstream::io_rstrm::acceptor::control_callbacks callbacks; + callbacks.m_on_status_cb = [&](const rstream::io_rstrm::status_extd& status) { + if (status.m_status && status.m_status.value() == "online") { + ++online_status_count; + if (online_status_count == 2) { + second_tunnel_online.store(true, std::memory_order_release); + } + } + else if (status.m_status && status.m_status.value().starts_with("disconnected")) { + ++disconnected_count; + } + }; + boost::system::error_code callback_error; + network_acceptor.set_control_callbacks(callbacks, callback_error); + check(!callback_error, "failed to install reconnecting acceptor callbacks"); + rstream::io_rstrm::socket peer(io_context.get_executor()); + rstream::io_rstrm::endpoint accepted_endpoint; + auto request = std::make_shared>(request_payload.size()); + bool stream_replied = false; + watchdog test_watchdog(io_context); + network_acceptor.async_accept(peer, accepted_endpoint, [&](const boost::system::error_code& accept_error) { + check(!accept_error, "acceptor failed to establish the stream before control loss"); + boost::asio::async_read(peer, boost::asio::buffer(*request), [&](const boost::system::error_code& read_error, std::size_t read) { + check(!read_error, "accepted stream read failed after acceptor reconnection"); + check(read == request->size(), "accepted stream read was truncated after acceptor reconnection"); + check(std::string(request->data(), request->size()) == request_payload, "accepted stream payload changed after acceptor reconnection"); + auto reply = std::make_shared(reply_payload); + boost::asio::async_write(peer, boost::asio::buffer(*reply), [&, reply](const boost::system::error_code& write_error, std::size_t written) { + check(!write_error, "accepted stream write failed after acceptor reconnection"); + check(written == reply->size(), "accepted stream reply was truncated after acceptor reconnection"); + stream_replied = true; + boost::system::error_code close_error; + peer.close(close_error); + check(!close_error, "failed to close accepted stream after reconnect test"); + network_acceptor.close(close_error); + check(!close_error, "failed to close reconnecting acceptor"); + }); + }); + }); + io_context.run(); + test_watchdog.complete(); + engine.join(); + check(!test_watchdog.timed_out(), "acceptor reconnect check timed out"); + check(disconnected_count == 1, "acceptor reported " + std::to_string(disconnected_count) + " liveness disconnections instead of one"); + check(online_status_count == 2, "acceptor reported " + std::to_string(online_status_count) + " online states instead of two"); + check(stream_replied, "accepted stream did not survive acceptor reconnection"); +} + static void check_client_rejects_invalid_liveness_acknowledgement() { fake_engine engine; @@ -2968,6 +3090,7 @@ int main(int argc, char** argv) run_selected_check(selected, "client_expires_missing_liveness_acknowledgement", check_client_expires_missing_liveness_acknowledgement); run_selected_check(selected, "client_tolerates_intermittent_heartbeat_loss", check_client_tolerates_intermittent_heartbeat_loss); run_selected_check(selected, "established_stream_survives_control_liveness_timeout", check_established_stream_survives_control_liveness_timeout); + run_selected_check(selected, "acceptor_reconnects_after_liveness_timeout_without_breaking_established_stream", check_acceptor_reconnects_after_liveness_timeout_without_breaking_established_stream); run_selected_check(selected, "client_rejects_invalid_liveness_acknowledgement", check_client_rejects_invalid_liveness_acknowledgement); run_selected_check(selected, "client_rejects_replayed_liveness_acknowledgement", check_client_rejects_replayed_liveness_acknowledgement); run_selected_check(selected, "client_rejects_invalid_liveness_configuration", check_client_rejects_invalid_liveness_configuration); From 942b1261983953a43f711d30c385be88b369038e Mon Sep 17 00:00:00 2001 From: uartnet <140632163+uartnet@users.noreply.github.com> Date: Sat, 15 Aug 2026 08:43:55 +0200 Subject: [PATCH 8/9] test(runtime): preserve coalesced proxy payloads --- test/io/io-rstrm/test_io_rstrm_handshake.cpp | 54 ++++++++++++++++++++ 1 file changed, 54 insertions(+) diff --git a/test/io/io-rstrm/test_io_rstrm_handshake.cpp b/test/io/io-rstrm/test_io_rstrm_handshake.cpp index 8f00b58..f575b0c 100644 --- a/test/io/io-rstrm/test_io_rstrm_handshake.cpp +++ b/test/io/io-rstrm/test_io_rstrm_handshake.cpp @@ -1,10 +1,14 @@ // See LICENSE file in the project root for license information. +#include #include #include #include +#include #include +#include #include +#include #include #include @@ -13,7 +17,9 @@ #include #include #include +#include #include +#include #include #include @@ -264,6 +270,53 @@ static void check_proxy_success_response_completes() assert(server_called); } +static void check_proxy_response_preserves_coalesced_payload() +{ + boost::asio::io_context io_context; + auto socket_a = std::make_shared(io_context.get_executor()); + auto socket_b = std::make_shared(io_context.get_executor()); + rstream::test::connect_stream_pair(*socket_a, *socket_b); + test_stream stream(*socket_a, false); + rstream::io_rstrm::config config; + config.m_no_token = true; + config.m_zero_rtt = false; + bool handshake_called = false; + bool payload_called = false; + std::array received{}; + handshake_type handshake(stream, rstream::io::make_address("engine.example:443"), config); + handshake.async_run(handshake_type::type::proxy_req, "stream-123", boost::none, [&](const boost::system::error_code& error_code) { + handshake_called = true; + assert(!error_code); + boost::asio::async_read(*socket_a, boost::asio::buffer(received), [&](const boost::system::error_code& read_error, std::size_t size) { + assert(!read_error); + assert(size == received.size()); + payload_called = true; + }); + }); + boost::asio::co_spawn(io_context.get_executor(), [socket = socket_b]() -> boost::asio::awaitable { + payloader_type payloader(*socket); + auto request = rstream::core::make_buffer_allocated(4096); + co_await payloader.async_recv(request, boost::asio::use_awaitable); + protobuf::Message response; + response.mutable_proxy_rsp(); + auto payload = serialize_message(response); + const std::string application_payload = "world"; + const auto payload_size = static_cast(payload.get_size()); + std::vector wire(sizeof(payload_size) + payload_size + application_payload.size()); + wire[0] = static_cast(payload_size >> 24); + wire[1] = static_cast(payload_size >> 16); + wire[2] = static_cast(payload_size >> 8); + wire[3] = static_cast(payload_size); + std::memcpy(wire.data() + sizeof(payload_size), payload.map().get_const_data(), payload_size); + std::memcpy(wire.data() + sizeof(payload_size) + payload_size, application_payload.data(), application_payload.size()); + co_await boost::asio::async_write(*socket, boost::asio::buffer(wire), boost::asio::use_awaitable); + co_return; }, boost::asio::detached); + io_context.run(); + assert(handshake_called); + assert(payload_called); + assert(std::string(received.data(), received.size()) == "world"); +} + #ifdef RSTREAM_WITH_IO_STREAMS static void check_proxy_secret_is_allowed_with_mtls_agent_auth() { @@ -439,6 +492,7 @@ int main(int argc, char** argv) check_stream_response_error_is_mapped(); check_stream_success_response_completes(); check_proxy_success_response_completes(); + check_proxy_response_preserves_coalesced_payload(); #ifdef RSTREAM_WITH_IO_STREAMS check_proxy_secret_is_allowed_with_mtls_agent_auth(); #endif From 1edeabb7cafd8ca6fca64321b5f9dcb497978665 Mon Sep 17 00:00:00 2001 From: uartnet <140632163+uartnet@users.noreply.github.com> Date: Sun, 16 Aug 2026 16:34:58 +0200 Subject: [PATCH 9/9] docs(tls): record Windows half-close audit gap --- TODO.md | 1 + 1 file changed, 1 insertion(+) diff --git a/TODO.md b/TODO.md index 6cc093a..0ff2df7 100644 --- a/TODO.md +++ b/TODO.md @@ -6,6 +6,7 @@ The focused multithreaded TLS half-close test passes with the quality, AddressSa This area is deliberately not the critical path of the current guide and sample validation. Before declaring the TLS lifecycle audit exhaustive, complete the following follow-up matrix: +- Diagnose the reproducible Windows shared-library timeout in `check_tls_half_close_preserves_receive_direction`: both the initial CI run and its isolated rerun reached the 10-second deadline at `test_io_common_stream_tls.cpp:861` and then aborted with `0xc0000409`, while the Windows static build and every Linux/macOS variant passed. Replace the terminal assertion with operation-level timeout evidence before changing behavior, identify which half-close completion is missing, and prove the correction through repeated Windows shared-library runs. Do not classify this as an OpenSSL defect without that evidence. - Exercise TLS 1.2 and TLS 1.3 handshake, cancellation, half-close, peer-close, timeout, and abrupt-reset paths with one and several `io_context` worker threads. - Repeat client/server interoperability in both directions for Go and C++, including concurrent streams and shutdown during backpressure. - Run long-duration and high-concurrency stress tests under AddressSanitizer and ThreadSanitizer, and retain machine-readable evidence in CI.