feat(runtime): harden control liveness and stream lifecycles

TODO.md

@@ -0,0 +1,16 @@
+# 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:
+
+- 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.
+- 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.

bin/ncat/bin/main.cpp

@@ -3,6 +3,7 @@
 #include <cmath>
 #include <iostream>
 #include <thread>
+#include <vector>
 
 #include <boost/asio/io_context.hpp>
 #include <boost/asio/signal_set.hpp>
@@ -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<std::string> normalize_cli_arguments(int argc, char** argv)
+{
+  std::vector<std::string> 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");

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 {

bin/ncat/lib/rstream/ncat/server.cpp

@@ -44,6 +44,13 @@
 #include <boost/process/child.hpp>
 #include <boost/process/io.hpp>
 #endif
+#ifndef _WIN32
+#if __has_include(<boost/process/v1/posix.hpp>)
+#include <boost/process/v1/posix.hpp>
+#else
+#include <boost/process/posix.hpp>
+#endif
+#endif
 #include <boost/signals2.hpp>
 #include <boost/system/errc.hpp>
 
@@ -52,6 +59,8 @@
 #ifdef _WIN32
 #include <shellapi.h>
 #include <windows.h>
+#else
+#include <unistd.h>
 #endif
 // clang-format on
 
@@ -64,6 +73,9 @@
 #include <rstream/core/log.hpp>
 #include <rstream/core/memory.hpp>
 #include <rstream/core/object_id.hpp>
+#ifndef _WIN32
+#include <rstream/core/posix/child_stdin.hpp>
+#endif
 #include <rstream/core/system.hpp>
 #ifdef RSTREAM_WITH_IO_STREAMS
 #include <rstream/io/detail/stream/async_connect.hpp>
@@ -389,7 +401,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<session_exec> {
  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,15 +466,17 @@ 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;
 
   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;
 
@@ -726,7 +740,7 @@ void server::impl::on_accept(const boost::system::error_code& error_code)
         session_ptr = std::make_shared<session_proxy>(std::move(m_socket), m_settings, session_id, boost::get<io::address>(m_config.m_remote));
       }
       else if (m_config.m_remote.type() == typeid(exec)) {
-        session_ptr = std::make_shared<session_exec>(std::move(m_socket), m_settings, session_id, boost::get<exec>(m_config.m_remote), m_config.m_local.m_url.scheme() == "tcp");
+        session_ptr = std::make_shared<session_exec>(std::move(m_socket), m_settings, session_id, boost::get<exec>(m_config.m_remote));
       }
       if (session_ptr) {
         m_sessions.insert(std::make_pair(session_id, session_ptr));
@@ -1172,17 +1186,20 @@ 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),
+#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),
@@ -1287,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<boost::process::child>(shell,
                                                         boost::process::args(args),
+#ifdef _WIN32
                                                         boost::process::std_in<m_child_stdin,
                                                                                boost::process::std_out>
                                                             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));
@@ -1317,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<boost::process::child>(exe,
                                                           boost::process::args(args),
+#ifdef _WIN32
                                                           boost::process::std_in<m_child_stdin,
                                                                                  boost::process::std_out>
                                                               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));
@@ -1329,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);
@@ -1409,10 +1439,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);
@@ -1444,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)

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<core::buffer> 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);
+    }
   }
 }
 

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: