pr comments

cpp/implementations/plain/incoming_call_handler.h

@@ -32,11 +32,25 @@ public:
 
     virtual void onMethods(const MethodsMessage& req, MethodsReply reply) = 0;
 
-    // `sink` stays alive until onUnsubscribe fires or the connection dies.
-    // The handler must call `sink(evt)` on every matching emission.
-    virtual void onSubscribe(const SubscribeMessage& req, EventSink sink) = 0;
+    // `sink` stays alive until onUnsubscribe fires or the connection
+    // dies. The handler must call `sink(evt)` on every matching emission.
+    //
+    // `connectionId` is an opaque per-connection token (the rpc layer
+    // passes the connection's `this` pointer). The handler keys sinks
+    // by it so a subsequent onUnsubscribe / onConnectionClosed can
+    // remove only the sinks belonging to that connection — sub/unsub
+    // frames don't carry a subscriber identifier on the wire.
+    virtual void onSubscribe(const SubscribeMessage& req, EventSink sink,
+                             const void* connectionId) = 0;
 
-    virtual void onUnsubscribe(const UnsubscribeMessage& req) = 0;
+    virtual void onUnsubscribe(const UnsubscribeMessage& req,
+                               const void* connectionId) = 0;
+
+    // Called when a connection is torn down (graceful close or error)
+    // so the handler can drop any sinks still keyed to it. Without
+    // this, a dropped client leaks subscriptions and the host keeps
+    // fanning events into dead sinks.
+    virtual void onConnectionClosed(const void* connectionId) = 0;
 
     virtual void onToken(const TokenMessage& req) = 0;
 };

cpp/implementations/plain/json_mapping.cpp

@@ -1,5 +1,7 @@
 #include "json_mapping.h"
 
+#include <type_traits>
+
 namespace logos::plain {
 
 using json = nlohmann::json;

cpp/implementations/plain/plain_logos_object.cpp

@@ -2,10 +2,14 @@
 
 #include "qvariant_rpc_value.h"
 
+#include <QCoreApplication>
 #include <QDebug>
+#include <QMetaObject>
+#include <QTimer>
 
 #include <chrono>
 #include <future>
+#include <thread>
 #include <utility>
 
 namespace logos::plain {
@@ -51,6 +55,32 @@ QVariant PlainLogosObject::callMethod(const QString& authToken,
     return rpcValueToQVariant(res.value);
 }
 
+namespace {
+
+// Hand `callback(result)` over to the Qt event loop so PlainLogosObject's
+// async path matches LogosObject's interface contract: callbacks are
+// always delivered on a subsequent event-loop iteration, on the Qt
+// thread, never synchronously and never racing with QObjects/UI code.
+//
+// Using QCoreApplication::instance() as the anchor means the queued
+// invocation lands on whichever thread runs the Qt event loop in this
+// process, regardless of which worker thread completed the future.
+// If the application has shut down (instance() is null), we drop the
+// callback rather than invoke it from an arbitrary thread.
+void postToQtEventLoop(PlainLogosObject::AsyncResultCallback callback,
+                       QVariant result)
+{
+    QCoreApplication* app = QCoreApplication::instance();
+    if (!app) return;
+    QMetaObject::invokeMethod(app,
+        [callback = std::move(callback), result = std::move(result)]() mutable {
+            callback(result);
+        },
+        Qt::QueuedConnection);
+}
+
+} // anonymous namespace
+
 void PlainLogosObject::callMethodAsync(const QString& authToken,
                                        const QString& methodName,
                                        const QVariantList& args,
@@ -59,7 +89,9 @@ void PlainLogosObject::callMethodAsync(const QString& authToken,
 {
     if (!callback) return;
     if (!m_conn || !m_conn->isOpen()) {
-        callback(QVariant());
+        // Defer even the failure path — LogosObject's contract requires
+        // callbacks on a subsequent event-loop iteration, never inline.
+        postToQtEventLoop(std::move(callback), QVariant());
         return;
     }
 
@@ -73,17 +105,20 @@ void PlainLogosObject::callMethodAsync(const QString& authToken,
     auto fut = std::make_shared<std::future<ResultMessage>>(
         m_conn->sendCall(std::move(msg)));
 
-    // Simple waiter thread. Acceptable here because callMethodAsync is
-    // already an infrequent path and spawns no-ops if the caller never
-    // awaits. A future iteration can fold this into the io_context.
+    // Waiter thread is per-call but the callback hops back to the Qt
+    // event loop before running, so it never races with Qt objects. A
+    // 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.
     std::thread([fut, timeoutMs, callback = std::move(callback)]() mutable {
         if (fut->wait_for(std::chrono::milliseconds(timeoutMs))
             != std::future_status::ready) {
-            callback(QVariant());
+            postToQtEventLoop(std::move(callback), QVariant());
             return;
         }
         auto res = fut->get();
-        callback(res.ok ? rpcValueToQVariant(res.value) : QVariant());
+        QVariant value = res.ok ? rpcValueToQVariant(res.value) : QVariant();
+        postToQtEventLoop(std::move(callback), std::move(value));
     }).detach();
 }
 

cpp/implementations/plain/plain_logos_object.h

@@ -8,6 +8,8 @@
 #include <memory>
 #include <mutex>
 #include <string>
+#include <utility>
+#include <vector>
 
 namespace logos::plain {
 

cpp/implementations/plain/plain_transport_connection.cpp

@@ -11,8 +11,11 @@
 #include <boost/asio/connect.hpp>
 #include <boost/asio/ip/tcp.hpp>
 #include <boost/asio/ssl/context.hpp>
+#include <boost/asio/ssl/host_name_verification.hpp>
 #include <boost/asio/ssl/stream.hpp>
 
+#include <openssl/ssl.h>
+
 namespace logos::plain {
 
 namespace {
@@ -77,6 +80,25 @@ bool PlainTransportConnection::connectToHost()
         if (m_cfg.protocol == LogosProtocol::TcpSsl) {
             auto ctx = buildClientSslCtx(m_cfg);
             SslStream stream(ioc, ctx);
+            // Set SNI: TLS clients must advertise the target host name
+            // in the ClientHello so the server picks the right cert
+            // (and so any intermediate proxy can route correctly). Without
+            // this, vhost-style deployments would terminate the handshake.
+            // Cast through the OpenSSL macro because Asio doesn't expose
+            // SNI configuration at the wrapper level.
+            if (!SSL_set_tlsext_host_name(stream.native_handle(),
+                                          m_cfg.host.c_str())) {
+                qWarning() << "PlainTransportConnection: SSL_set_tlsext_host_name failed";
+            }
+            // Verify the peer's certificate name matches the host we
+            // dialed when verifyPeer is on. verify_peer alone only
+            // validates the chain — without host-name verification a
+            // valid cert for a *different* name would still pass, which
+            // is exactly the MITM hole verify_peer is meant to close.
+            if (m_cfg.verifyPeer) {
+                stream.set_verify_callback(
+                    boost::asio::ssl::host_name_verification(m_cfg.host));
+            }
             boost::asio::connect(stream.lowest_layer(), endpoints);
             stream.handshake(boost::asio::ssl::stream_base::client);
             auto conn = std::make_shared<SslConnection>(

cpp/implementations/plain/plain_transport_host.cpp

@@ -10,6 +10,8 @@
 #include <QDebug>
 #include <QMetaObject>
 
+#include <atomic>
+
 #include <boost/asio/ssl/context.hpp>
 #include <boost/version.hpp>
 #include <openssl/ssl.h>
@@ -377,25 +379,49 @@ void PlainTransportHost::onMethods(const MethodsMessage& req, MethodsReply reply
     }, Qt::QueuedConnection);
 }
 
-void PlainTransportHost::onSubscribe(const SubscribeMessage& req, EventSink sink)
+void PlainTransportHost::onSubscribe(const SubscribeMessage& req, EventSink sink,
+                                     const void* connectionId)
 {
     std::lock_guard<std::mutex> g(m_mu);
     auto it = m_published.find(req.object);
     if (it == m_published.end()) return;
-    // Sinks are keyed by functor-shared-pointer address. The caller (RpcConnection)
-    // moves the sink in; we key by its move-captured lambda-object's address,
-    // which is unique per subscription.
-    static std::atomic<uint64_t> counter{0};
-    it->second.sinksByEvent[req.eventName][
-        reinterpret_cast<const void*>(counter.fetch_add(1) + 1)] = std::move(sink);
+    // Sinks are keyed by the originating connection so that
+    // onUnsubscribe / onConnectionClosed can remove only sinks
+    // belonging to that connection — sub/unsub frames don't carry a
+    // subscriber id on the wire.
+    it->second.sinksByEvent[req.eventName][connectionId] = std::move(sink);
 }
 
-void PlainTransportHost::onUnsubscribe(const UnsubscribeMessage& req)
+void PlainTransportHost::onUnsubscribe(const UnsubscribeMessage& req,
+                                       const void* connectionId)
 {
     std::lock_guard<std::mutex> g(m_mu);
     auto it = m_published.find(req.object);
     if (it == m_published.end()) return;
-    it->second.sinksByEvent.erase(req.eventName);
+    auto evtIt = it->second.sinksByEvent.find(req.eventName);
+    if (evtIt == it->second.sinksByEvent.end()) return;
+    // Only drop the requesting connection's sink — other clients
+    // subscribed to the same (object, event) keep theirs. Previously
+    // this erased the entire eventName entry, taking every other
+    // subscriber down with it.
+    evtIt->second.erase(connectionId);
+    if (evtIt->second.empty()) it->second.sinksByEvent.erase(evtIt);
+}
+
+void PlainTransportHost::onConnectionClosed(const void* connectionId)
+{
+    std::lock_guard<std::mutex> g(m_mu);
+    // Sweep every published object's per-event sink table and drop any
+    // entries belonging to the closed connection. Without this, a
+    // crashing/disconnecting client (which never sends Unsubscribe)
+    // leaves dead sinks in the table forever.
+    for (auto& [_name, pub] : m_published) {
+        for (auto evtIt = pub.sinksByEvent.begin(); evtIt != pub.sinksByEvent.end(); ) {
+            evtIt->second.erase(connectionId);
+            if (evtIt->second.empty()) evtIt = pub.sinksByEvent.erase(evtIt);
+            else ++evtIt;
+        }
+    }
 }
 
 void PlainTransportHost::onToken(const TokenMessage& req)

cpp/implementations/plain/plain_transport_host.h

@@ -49,8 +49,11 @@ public:
     // IncomingCallHandler
     void onCall(const CallMessage& req, CallReply reply) override;
     void onMethods(const MethodsMessage& req, MethodsReply reply) override;
-    void onSubscribe(const SubscribeMessage& req, EventSink sink) override;
-    void onUnsubscribe(const UnsubscribeMessage& req) override;
+    void onSubscribe(const SubscribeMessage& req, EventSink sink,
+                     const void* connectionId) override;
+    void onUnsubscribe(const UnsubscribeMessage& req,
+                       const void* connectionId) override;
+    void onConnectionClosed(const void* connectionId) override;
     void onToken(const TokenMessage& req) override;
 
     // Internal: deliver an event emitted by the wrapped QObject to every

cpp/implementations/plain/qvariant_rpc_value.cpp

@@ -4,6 +4,9 @@
 
 #include <QMetaType>
 
+#include <cmath>
+#include <limits>
+
 namespace logos::plain {
 
 namespace {

cpp/implementations/plain/rpc_connection.h

@@ -264,13 +264,19 @@ void RpcConnection<Stream>::dispatchIncoming(AnyMessage msg)
 
         } else if constexpr (std::is_same_v<T, SubscribeMessage>) {
             if (!m_handler) return;
-            auto self = this->shared_from_this();
-            m_handler->onSubscribe(m, [self](EventMessage evt) {
-                self->sendEvent(std::move(evt));
-            });
+            // weak_ptr capture so the host's stored sink doesn't keep the
+            // connection alive past its natural lifetime — without this,
+            // `[self]` would leak every subscribed connection until
+            // unsubscribe (which a crashing client never sends).
+            std::weak_ptr<RpcConnection<Stream>> weak = this->shared_from_this();
+            const void* connId = static_cast<const void*>(this);
+            m_handler->onSubscribe(m, [weak](EventMessage evt) {
+                if (auto self = weak.lock()) self->sendEvent(std::move(evt));
+            }, connId);
 
         } else if constexpr (std::is_same_v<T, UnsubscribeMessage>) {
-            if (m_handler) m_handler->onUnsubscribe(m);
+            if (m_handler)
+                m_handler->onUnsubscribe(m, static_cast<const void*>(this));
 
         } else if constexpr (std::is_same_v<T, TokenMessage>) {
             if (m_handler) m_handler->onToken(m);
@@ -420,6 +426,14 @@ void RpcConnection<Stream>::fail(const std::string& reason)
         m_stream.lowest_layer().close(ignore);
     } catch (...) {}
 
+    // Notify the dispatch handler so it can drop any subscriptions still
+    // keyed to this connection. Without this, a connection that drops
+    // without sending Unsubscribe leaks sinks in the host's per-event map.
+    if (m_handler) {
+        try { m_handler->onConnectionClosed(static_cast<const void*>(this)); }
+        catch (...) {}
+    }
+
     if (errCb) errCb(reason);
 }
 

cpp/implementations/plain/rpc_message.cpp

@@ -1,5 +1,7 @@
 #include "rpc_message.h"
 
+#include <type_traits>
+
 namespace logos::plain {
 
 MessageType messageTypeOf(const AnyMessage& m)