fix(plain): close the RPC acceptor on the server's strand, not the caller's thread (#39)

RpcServerTcp::stop() and RpcServerSsl::stop() closed m_acceptor on whatever
thread called them — in practice the host thread, via ~PlainTransportHost —
while doAccept() re-armed async_accept from inside its own completion handler,
on the io worker. Nothing serialized the two.

This is the acceptor half of the race PR #38 fixed for RpcConnection, and it
fails identically: asio acceptors are "Shared objects: Unsafe", and close()
runs cleanup_descriptor_data(), which nulls the reactor's per-descriptor state
while reactive_socket_service_base::start_op() holds it by reference. It was
left out of #38 because every backtrace captured in the wild was a write
initiation, never an accept — but it reproduces on demand:

  EXC_BAD_ACCESS  KERN_INVALID_ADDRESS at 0x98
    logos::plain::RpcServerTcp::doAccept()
    ...reactive_socket_move_accept_op<...>::do_complete(...)
    logos::plain::IoContextPool::IoContextPool()::$_0    <- io worker thread

Both servers now own a strand. doAccept()'s completion handler is
bind_executor'd onto it (so the re-arm runs there) and stop() hands the close
to it with dispatch() — inline when already on the strand, queued and
non-blocking from anywhere else, exactly as RpcConnection::closeStreamOnStrand
does.

start() still runs open/bind/listen inline: callers read boundPort() the moment
it returns. That is safe because no async op on the acceptor exists yet, and
PlainTransportHost serializes start()/stop() under its own mutex. Only the
accept loop moves onto the strand, which is invisible to clients — listen() has
already run, so an early connect waits in the backlog.

Deferring the close leaves the listener open for the microseconds between
stop() returning and the strand running it, so a connection can still be
accepted in that gap. The accept path therefore tests m_stopped and publishes
the connection under one lock, and drops a late socket instead of wrapping it
in a connection and stop()ing it — conn->stop() would call
onConnectionClosed() on the IncomingCallHandler whose destructor started this
teardown. The TLS server gets the same guard, where it was already latent: an
async_handshake in flight was never aborted by closing the acceptor.

Adds RpcServerTeardownTest: a start/connect/stop stress loop shaped like
test_rpc_connection_teardown.cpp, plus a round-trip check that a client
connecting the instant start() returns is still served.

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>

cpp/implementations/plain/rpc_server.cpp

@@ -1,5 +1,7 @@
 #include "rpc_server.h"
 
+#include <boost/asio/bind_executor.hpp>
+#include <boost/asio/dispatch.hpp>
 #include <boost/asio/ip/address.hpp>
 
 #include <QDebug>
@@ -16,6 +18,7 @@ RpcServerTcp::RpcServerTcp(boost::asio::io_context& ioc,
                            std::shared_ptr<IWireCodec> codec,
                            IncomingCallHandler* handler)
     : m_acceptor(ioc)
+    , m_strand(boost::asio::make_strand(m_acceptor.get_executor()))
     , m_codec(std::move(codec))
     , m_handler(handler)
     , m_host(host)
@@ -31,6 +34,12 @@ bool RpcServerTcp::start()
         m_port);
     if (ec) return false;
 
+    // open/bind/listen run here, on the caller's thread, rather than on the
+    // strand: start() has to be synchronous through them because callers read
+    // boundPort() the moment it returns (port=0 means "let the kernel pick").
+    // That is safe because no async operation on m_acceptor exists yet — the
+    // first one is armed below — and callers must not overlap start() with
+    // stop() (PlainTransportHost holds its own mutex across both).
     m_acceptor.open(ep.protocol(), ec);       if (ec) return false;
     m_acceptor.set_option(boost::asio::socket_base::reuse_address(true), ec);
     m_acceptor.bind(ep, ec);                  if (ec) return false;
@@ -38,7 +47,13 @@ bool RpcServerTcp::start()
     if (ec) return false;
 
     m_boundPort = m_acceptor.local_endpoint().port();
-    doAccept();
+
+    // From here on every touch of m_acceptor goes through the strand. Arming
+    // the first accept there rather than inline is invisible to clients:
+    // listen() has already run, so anything that connects before the strand
+    // gets its turn waits in the backlog.
+    auto self = shared_from_this();
+    boost::asio::dispatch(m_strand, [self] { self->doAccept(); });
     return true;
 }
 
@@ -55,25 +70,73 @@ void RpcServerTcp::stop()
     {
         std::lock_guard<std::mutex> g(m_mu);
         m_stopped = true;
-        boost::system::error_code ignore;
-        m_acceptor.close(ignore);
         conns.swap(m_conns);
     }
+    closeAcceptorOnStrand();
     for (auto& c : conns) c->stop("server stopped");
 }
 
+void RpcServerTcp::closeAcceptor()
+{
+    boost::system::error_code ignore;
+    m_acceptor.close(ignore);
+}
+
+void RpcServerTcp::closeAcceptorOnStrand()
+{
+    // The acceptor is subject to exactly the race RpcConnection's stream was
+    // (see closeStreamOnStrand() there): asio acceptors are documented "Shared
+    // objects: Unsafe", and close() runs cleanup_descriptor_data(), which nulls
+    // the reactor's per-descriptor state. doAccept() re-arms async_accept from
+    // inside its own completion handler — on the io thread — while stop() is
+    // called from an arbitrary caller thread (in practice the host thread, via
+    // ~PlainTransportHost). Closing there let the reactor dereference the
+    // descriptor state the close had just nulled, inside
+    // reactive_socket_service_base::start_op(): SIGSEGV at +0x98 on the io
+    // thread, with RpcServerTcp::doAccept() reached from the accept completion
+    // handler at the top of the backtrace.
+    //
+    // dispatch() (not post()) for the same reason as in RpcConnection: when the
+    // caller is already on the strand it closes inline; from any other thread it
+    // queues and returns immediately, so teardown never blocks. The lambda holds
+    // a shared_ptr, so a close queued from a destructor still finds a live
+    // object; if the io_context stops first, the acceptor is closed by
+    // ~basic_socket_acceptor when the server is destroyed.
+    std::shared_ptr<RpcServerTcp> self;
+    try { self = shared_from_this(); } catch (...) {}
+    if (!self) {
+        // No owning shared_ptr — the object is mid-destruction, so no other
+        // thread can still be holding it to run an acceptor operation.
+        closeAcceptor();
+        return;
+    }
+    boost::asio::dispatch(m_strand, [self] { self->closeAcceptor(); });
+}
+
 void RpcServerTcp::doAccept()
 {
     auto self = shared_from_this();
     m_acceptor.async_accept(
+        boost::asio::bind_executor(m_strand,
         [self](const boost::system::error_code& ec,
                boost::asio::ip::tcp::socket socket) {
             if (ec) return;  // acceptor probably closed; quietly exit.
-            auto conn = std::make_shared<TcpConnection>(
-                std::move(socket), self->m_codec, self->m_handler);
+            std::shared_ptr<TcpConnection> conn;
             {
                 std::lock_guard<std::mutex> g(self->m_mu);
-                if (self->m_stopped) { conn->stop("server stopped"); return; }
+                // Test and publish under one lock. A socket accepted after
+                // stop() is dropped here — closed by `socket`'s destructor —
+                // rather than wrapped in a connection and stop()ed: stop()
+                // would reach IncomingCallHandler::onConnectionClosed, and the
+                // handler is the thing that calls RpcServer::stop() from its
+                // own destructor (~PlainTransportHost), so by the time this
+                // runs it may already be gone. Deferring the acceptor close
+                // onto the strand widened that window from "impossible"
+                // (close() aborted the pending accept before stop() returned)
+                // to "a few microseconds", which is long enough to matter.
+                if (self->m_stopped) return;
+                conn = std::make_shared<TcpConnection>(
+                    std::move(socket), self->m_codec, self->m_handler);
                 self->m_conns.push_back(conn);
             }
             std::weak_ptr<RpcServerTcp> weakSelf = self;
@@ -87,7 +150,7 @@ void RpcServerTcp::doAccept()
             });
             conn->start();
             self->doAccept();
-        });
+        }));
 }
 
 // ── RpcServerSsl ──────────────────────────────────────────────────────────
@@ -99,6 +162,7 @@ RpcServerSsl::RpcServerSsl(boost::asio::io_context& ioc,
                            std::shared_ptr<IWireCodec> codec,
                            IncomingCallHandler* handler)
     : m_acceptor(ioc)
+    , m_strand(boost::asio::make_strand(m_acceptor.get_executor()))
     , m_sslCtx(std::move(sslCtx))
     , m_codec(std::move(codec))
     , m_handler(handler)
@@ -109,6 +173,8 @@ RpcServerSsl::RpcServerSsl(boost::asio::io_context& ioc,
 
 bool RpcServerSsl::start()
 {
+    // See RpcServerTcp::start() for why bind/listen stay on the caller's thread
+    // and only the accept loop moves onto the strand.
     boost::system::error_code ec;
     boost::asio::ip::tcp::endpoint ep(
         boost::asio::ip::make_address(m_host, ec),
@@ -122,7 +188,9 @@ bool RpcServerSsl::start()
     if (ec) return false;
 
     m_boundPort = m_acceptor.local_endpoint().port();
-    doAccept();
+
+    auto self = shared_from_this();
+    boost::asio::dispatch(m_strand, [self] { self->doAccept(); });
     return true;
 }
 
@@ -134,21 +202,42 @@ void RpcServerSsl::stop()
     {
         std::lock_guard<std::mutex> g(m_mu);
         m_stopped = true;
-        boost::system::error_code ignore;
-        m_acceptor.close(ignore);
         conns.swap(m_conns);
     }
+    closeAcceptorOnStrand();
     for (auto& c : conns) c->stop("server stopped");
 }
 
+void RpcServerSsl::closeAcceptor()
+{
+    boost::system::error_code ignore;
+    m_acceptor.close(ignore);
+}
+
+void RpcServerSsl::closeAcceptorOnStrand()
+{
+    // See RpcServerTcp::closeAcceptorOnStrand().
+    std::shared_ptr<RpcServerSsl> self;
+    try { self = shared_from_this(); } catch (...) {}
+    if (!self) {
+        closeAcceptor();
+        return;
+    }
+    boost::asio::dispatch(m_strand, [self] { self->closeAcceptor(); });
+}
+
 void RpcServerSsl::doAccept()
 {
     auto self = shared_from_this();
     m_acceptor.async_accept(
+        boost::asio::bind_executor(m_strand,
         [self](const boost::system::error_code& ec,
                boost::asio::ip::tcp::socket socket) {
             if (ec) return;
             auto stream = std::make_shared<SslStream>(std::move(socket), self->m_sslCtx);
+            // The handshake completes off the strand: it touches the new
+            // stream, never m_acceptor, and every server member it reads
+            // afterwards is guarded by m_mu.
             stream->async_handshake(
                 boost::asio::ssl::stream_base::server,
                 [self, stream](const boost::system::error_code& hs) {
@@ -171,11 +260,17 @@ void RpcServerSsl::doAccept()
                     // duration of async_handshake (so the buffer
                     // outlives the dispatch); now move the underlying
                     // stream into the connection by value.
-                    auto conn = std::make_shared<SslConnection>(
-                        std::move(*stream), self->m_codec, self->m_handler);
+                    std::shared_ptr<SslConnection> conn;
                     {
                         std::lock_guard<std::mutex> g(self->m_mu);
-                        if (self->m_stopped) { conn->stop("server stopped"); return; }
+                        // Dropped rather than stop()ed once the server is
+                        // stopped — see the matching note in
+                        // RpcServerTcp::doAccept(). Closing the acceptor never
+                        // aborted a handshake already in flight, so this path
+                        // has always had to survive a handler that is gone.
+                        if (self->m_stopped) return;
+                        conn = std::make_shared<SslConnection>(
+                            std::move(*stream), self->m_codec, self->m_handler);
                         self->m_conns.push_back(conn);
                     }
                     std::weak_ptr<RpcServerSsl> weakSelf = self;
@@ -190,7 +285,7 @@ void RpcServerSsl::doAccept()
                     conn->start();
                 });
             self->doAccept();
-        });
+        }));
 }
 
 } // namespace logos::plain