fix(protocol): stop the macOS flake that was sinking #41

Three real races the new CallErrorAfterAcquire suite exposed (and that
Copilot flagged on the QtRO half):

1. ~PlainTransportHost stopped the acceptor but did not quiesce the shared
   Asio I/O thread. Server-side RpcConnections hold a raw IncomingCallHandler*
   back to the host; a fail()/onConnectionClosed racing teardown freed the
   handler mid-call. That is the macOS CI SIGSEGV in
   AsyncSuccessStillReportsTheValue — it fires with no output of its own
   because the previous live-host test's destructor left the heap corrupted.
   Restore the I/O barrier that landed on the qtfree branches but never on
   master (proven: 80/80 clean on the CI crash sequence that was ~2/50 before).

2. PlainLogosObject::callMethodAsync detached its per-call waiter while
   capturing `this`. release()/delete this could then race the waiter.
   Join waiters in the destructor/release, and register the thread under the
   lock before it can outrun teardown.

3. QtRO async could deliver the user callback twice when the timeout timer
   and the pending-call watcher finished around the same moment, violating
   the exactly-once contract. Gate both paths (and the deferred-completion
   arm) on one atomic.

Also drain queued onCall invokes after host.reset() in the #40 live-target
control, matching LiveHost's teardown discipline.

Co-authored-by: Cursor <cursoragent@cursor.com>

cpp/implementations/plain/plain_logos_object.cpp

@@ -26,6 +26,20 @@ PlainLogosObject::PlainLogosObject(std::string objectName,
 PlainLogosObject::~PlainLogosObject()
 {
     disconnectEvents();
+    joinWaiters();
+}
+
+void PlainLogosObject::joinWaiters()
+{
+    std::vector<std::thread> waiters;
+    {
+        std::lock_guard<std::mutex> g(m_waiterMu);
+        waiters.swap(m_waiters);
+    }
+    for (auto& t : waiters) {
+        if (t.joinable())
+            t.join();
+    }
 }
 
 QVariant PlainLogosObject::callMethod(const QString& authToken,
@@ -215,34 +229,45 @@ void PlainLogosObject::callMethodAsyncWithError(const QString& authToken,
     // future iteration can fold this wait into the shared Asio
     // io_context (the connection already runs on it) so we don't spin
     // up a thread per pending RPC.
+    //
+    // The thread is JOINed in joinWaiters() (destructor / release), not
+    // detached: capturing `this` for awaitCompletion / m_objectName is
+    // only safe while the object is alive, and release() used to
+    // `delete this` while a waiter could still be mid-flight.
+    const std::string objectName = m_objectName;
     const std::string method = methodName.toStdString();
-    std::thread([this, fut, timeoutMs, methodName, method,
-                 callback = std::move(callback)]() mutable {
-        if (fut->wait_for(std::chrono::milliseconds(timeoutMs))
-            != std::future_status::ready) {
-            postToQtEventLoop(std::move(callback), QVariant(),
-                              logos::callErrorTimeout(m_objectName, method,
-                                                      timeoutMs));
-            return;
-        }
-        auto res = fut->get();
-        if (!res.ok) {
-            postToQtEventLoop(std::move(callback), QVariant(),
-                              logos::callErrorFromWire(m_objectName, res.errCode,
-                                                       res.err));
-            return;
-        }
-        QVariant value = rpcValueToQVariant(res.value);
-        // Resolve a "multi" provider's deferred completion (sentinel → wait for
-        // the completion event) right here on the waiter thread.
-        logos::CallError err;
-        {
-            QString callId;
-            if (logos::isPendingCallSentinel(value, &callId))
-                value = awaitCompletion(callId, timeoutMs, methodName, &err);
-        }
-        postToQtEventLoop(std::move(callback), std::move(value), std::move(err));
-    }).detach();
+    // Register under the lock BEFORE the thread can outrun release(): a
+    // detach-then-push left a window where delete this raced the waiter.
+    {
+        std::lock_guard<std::mutex> g(m_waiterMu);
+        m_waiters.emplace_back([this, objectName, fut, timeoutMs, methodName, method,
+                                callback = std::move(callback)]() mutable {
+            if (fut->wait_for(std::chrono::milliseconds(timeoutMs))
+                != std::future_status::ready) {
+                postToQtEventLoop(std::move(callback), QVariant(),
+                                  logos::callErrorTimeout(objectName, method,
+                                                          timeoutMs));
+                return;
+            }
+            auto res = fut->get();
+            if (!res.ok) {
+                postToQtEventLoop(std::move(callback), QVariant(),
+                                  logos::callErrorFromWire(objectName, res.errCode,
+                                                           res.err));
+                return;
+            }
+            QVariant value = rpcValueToQVariant(res.value);
+            // Resolve a "multi" provider's deferred completion (sentinel → wait for
+            // the completion event) right here on the waiter thread.
+            logos::CallError err;
+            {
+                QString callId;
+                if (logos::isPendingCallSentinel(value, &callId))
+                    value = awaitCompletion(callId, timeoutMs, methodName, &err);
+            }
+            postToQtEventLoop(std::move(callback), std::move(value), std::move(err));
+        });
+    }
 }
 
 bool PlainLogosObject::informModuleToken(const QString& authToken,
@@ -329,7 +354,12 @@ void PlainLogosObject::release()
     // the connection for every other holder too, so just unsubscribe our
     // own events and drop our reference — the connection stays alive
     // until PlainTransportConnection itself is destroyed.
+    //
+    // joinWaiters() before delete: in-flight async waiters capture `this`
+    // (for awaitCompletion). Detaching them used to let release() free the
+    // object under a still-running waiter.
     disconnectEvents();
+    joinWaiters();
     m_conn.reset();
     delete this;
 }

cpp/implementations/plain/plain_logos_object.h

@@ -10,6 +10,7 @@
 #include <memory>
 #include <mutex>
 #include <string>
+#include <thread>
 #include <utility>
 #include <vector>
 
@@ -83,6 +84,11 @@ private:
                              const QString& methodName = QString(),
                              logos::CallError* err = nullptr);
 
+    // Join every per-call waiter before tearing the object down. callMethodAsync
+    // used to detach those threads, so release()/delete racing an in-flight
+    // wait was a use-after-free on `this` (m_objectName, awaitCompletion, …).
+    void joinWaiters();
+
     std::string                          m_objectName;
     std::shared_ptr<RpcConnectionBase>   m_conn;
     std::mutex                           m_mu;
@@ -92,6 +98,9 @@ private:
     std::condition_variable              m_completionCv;
     std::map<QString, QVariant>          m_completions;
     bool                                 m_completionSubscribed = false;
+
+    std::mutex                           m_waiterMu;
+    std::vector<std::thread>             m_waiters;
 };
 
 } // namespace logos::plain

cpp/implementations/plain/plain_transport_host.cpp

@@ -11,7 +11,10 @@
 #include <QMetaObject>
 
 #include <atomic>
+#include <chrono>
+#include <future>
 
+#include <boost/asio/post.hpp>
 #include <boost/asio/ssl/context.hpp>
 #include <boost/version.hpp>
 #include <openssl/ssl.h>
@@ -210,6 +213,30 @@ PlainTransportHost::~PlainTransportHost()
     }
     if (tcp) tcp->stop();
     if (ssl) ssl->stop();
+
+    // Quiesce the I/O thread before this host (an IncomingCallHandler) is
+    // destroyed. Server-side RpcConnections hold a RAW `IncomingCallHandler*`
+    // back to us; a read completion racing this teardown runs
+    // RpcConnection::fail() on the I/O thread, which calls
+    // m_handler->onConnectionClosed(this). stop() above closes the sockets but
+    // does NOT wait for an already-executing fail() — so without this barrier
+    // the handler can be freed mid-call (a use-after-free that surfaced as a
+    // flaky SIGSEGV, including on macOS CI in CallErrorAfterAcquireTest right
+    // after a preceding live-host test tore its PlainTransportHost down).
+    // There is a single shared I/O thread, so a task posted now runs only after
+    // every in-flight/queued connection handler has completed; blocking on it
+    // guarantees no callback still references this host. Skip when we're ON
+    // the I/O thread (the in-flight handler is our own caller) to avoid
+    // self-deadlock.
+    if (tcp || ssl) {
+        auto& ioc = IoContextPool::shared().ioContext();
+        if (!ioc.get_executor().running_in_this_thread()) {
+            std::promise<void> drained;
+            auto fut = drained.get_future();
+            boost::asio::post(ioc, [&drained] { drained.set_value(); });
+            fut.wait_for(std::chrono::seconds(5)); // safety-bounded; work-guard keeps the thread alive
+        }
+    }
 }
 
 bool PlainTransportHost::start()

cpp/implementations/qt_remote/remote_transport.cpp

@@ -257,10 +257,34 @@ public:
         auto* timer = new QTimer(watcher);
         timer->setSingleShot(true);
 
+        // Exactly-once gate. The finished handler and the timeout timer can
+        // both be queued around the same moment; without a guard that races
+        // into a double callback, which violates callMethodAsyncWithError's
+        // contract and is a latent double-free for every consumer. The same
+        // gate wraps the deferred-completion path so a late completion cannot
+        // deliver after the initial timeout already has (or vice versa).
+        auto delivered = std::make_shared<std::atomic_bool>(false);
+        AsyncResultErrorCallback deliverOnce =
+            [delivered, callback = std::move(callback)](QVariant result,
+                                                       const logos::CallError& err) mutable {
+                if (delivered->exchange(true))
+                    return;
+                if (callback)
+                    callback(std::move(result), err);
+            };
+
         // Success handler -- delivers result on the consumer's thread
         QObject::connect(watcher, &QRemoteObjectPendingCallWatcher::finished,
-                         watcher, [this, callback, timer, timeoutMs, origin, method](QRemoteObjectPendingCallWatcher* w) {
+                         watcher, [this, deliverOnce, timer, timeoutMs, origin, method, delivered](QRemoteObjectPendingCallWatcher* w) {
             timer->stop(); // cancel timeout
+            // Timeout may already have won the race and deleteLater'd us; if
+            // the slot still runs, do not enter the deferred-completion path
+            // or we would arm a second delivery after the caller already saw
+            // a timeout.
+            if (delivered->load()) {
+                w->deleteLater();
+                return;
+            }
             QVariant result;
             logos::CallError err;
             if (w->error() == QRemoteObjectPendingCall::NoError) {
@@ -279,10 +303,10 @@ public:
                 QString callId;
                 if (logos::isPendingCallSentinel(result, &callId)) {
                     if (m_completions.contains(callId)) {
-                        callback(m_completions.take(callId), logos::CallError{});
+                        deliverOnce(m_completions.take(callId), logos::CallError{});
                         return;
                     }
-                    m_asyncCompletionCbs.insert(callId, callback);
+                    m_asyncCompletionCbs.insert(callId, deliverOnce);
                     // Bound the wait: a completion that never lands is a timeout,
                     // reported as one instead of as an empty result.
                     QTimer::singleShot(timeoutMs, m_helper, [this, callId, origin, method, timeoutMs]() {
@@ -296,13 +320,13 @@ public:
                     return;
                 }
             }
-            callback(result, err);
+            deliverOnce(result, err);
         }, Qt::QueuedConnection);
 
         // Timeout handler -- stops the watcher and reports the elapsed deadline
-        QObject::connect(timer, &QTimer::timeout, watcher, [watcher, callback, origin, method, timeoutMs]() {
+        QObject::connect(timer, &QTimer::timeout, watcher, [watcher, deliverOnce, origin, method, timeoutMs]() {
             qWarning() << "RemoteLogosObject: async callMethod timed out";
-            callback(QVariant(), logos::callErrorTimeout(origin, method, timeoutMs));
+            deliverOnce(QVariant(), logos::callErrorTimeout(origin, method, timeoutMs));
             watcher->deleteLater(); // also destroys the timer (child)
         });
 

tests/protocol/test_lp_invoke_async_error.cpp

@@ -207,4 +207,11 @@ TEST_F(LpInvokeAsyncErrorTest, LiveTargetStillReportsOkWithItsValue)
     // Let the deferred client teardown run before the host goes away.
     QCoreApplication::processEvents(QEventLoop::AllEvents, 50);
     host.reset();
+    // Drain any QueuedConnection onCall invokes that PlainTransportHost
+    // posted to `proxy` before stop() returned. Without this, those
+    // slots can fire after `proxy` is destroyed at scope exit — a UAF
+    // that corrupts the heap and segfaults the *next* test (seen on
+    // macOS as CallErrorAfterAcquireTest.AsyncSuccessStillReportsTheValue
+    // crashing with no output of its own).
+    QCoreApplication::processEvents(QEventLoop::AllEvents, 50);
 }