Dispatch rtc callbacks off the rtc threads; TS-parity full-node test scale

packages/streamr-dht/modules/connection/webrtc/WebrtcConnection.cppm

@@ -83,6 +83,13 @@ private:
     streamr::eventemitter::EventEmitter<WebrtcSignallingEvents>
         signallingEvents;
     std::recursive_mutex mMutex;
+    // Per-connection FIFO dispatch of the datachannel open/message emit
+    // chains: the rtc worker pool is fixed-size and also runs ICE/DTLS
+    // processing for every connection in the process, so the emit chains
+    // must not run inline there (the WebsocketConnection idiom; see its
+    // mDispatchExecutor).
+    streamr::utils::SharedSerialExecutor mDispatchExecutor{
+        streamr::utils::SharedExecutors::worker()};
 
     explicit WebrtcConnection(WebrtcConnectionParams&& params)
         : Connection(ConnectionType::WEBRTC), params(std::move(params)) {}
@@ -368,7 +375,14 @@ private:
                 defaultBufferThresholdLow));
         this->dataChannel->onOpen([self]() {
             SLogger::trace("dc.onOpened");
-            self->onDataChannelOpen();
+            // The Connected emit chain (handshaker, connection manager)
+            // must not run on the rtc worker thread — it starves the
+            // fixed-size pool that also processes ICE/DTLS handshakes
+            // (the same starvation the websocket poll thread had). The
+            // per-connection serial executor keeps Connected ordered
+            // before the Data emits below.
+            self->mDispatchExecutor.add(
+                [self]() { self->onDataChannelOpen(); });
         });
         this->dataChannel->onClosed([self]() {
             SLogger::trace("dc.closed");
@@ -382,7 +396,15 @@ private:
         this->dataChannel->onMessage([self](rtc::message_variant message) {
             SLogger::trace("dc.onMessage");
             if (std::holds_alternative<rtc::binary>(message)) {
-                self->emit<Data>(std::get<rtc::binary>(std::move(message)));
+                // Thin enqueue (see onOpen above): the rtc worker thread
+                // only copies the frame; the emit<Data> chain runs on the
+                // per-connection serial executor in arrival order.
+                self->mDispatchExecutor.add(
+                    [self,
+                     data =
+                         std::get<rtc::binary>(std::move(message))]() mutable {
+                        self->emit<Data>(std::move(data));
+                    });
             }
         });
     }

packages/streamr-dht/modules/connection/webrtc/WebrtcConnector.cppm

@@ -261,26 +261,78 @@ public:
         });
 
         // TS delFunc: the attempt is dropped once the connection settles
-        // either way (erasing an absent key is a no-op).
+        // either way. Every handler below erases the map entry ONLY if it
+        // still belongs to the connection/pending pair that fired — the map
+        // is keyed by nodeId and repeated connects to the same peer insert
+        // new pairs, so an unconditional erase(nodeId) from a stale handler
+        // (an older connection's Disconnected arriving late) deleted a
+        // freshly inserted attempt and orphaned its WebrtcConnection
+        // (observed: insert and stale erase within the same millisecond).
+        // An orphan is immortal — its rtc callbacks capture a strong self,
+        // broken only by doClose — and its dispatch executor's worker-pool
+        // KeepAlive then blocks process exit in the shared pool destructor
+        // (joinKeepAliveOnce). The captured raw pointers are compared for
+        // identity only, never dereferenced.
         connection->on<connectionevents::Disconnected>(
-            [this, nodeId](
+            [this, nodeId, connPtr = connection.get()](
                 bool /*gracefulLeave*/,
                 uint64_t /*code*/,
                 const std::string& /*reason*/) {
                 std::scoped_lock lock(this->mMutex);
-                this->ongoingConnectAttempts.erase(nodeId);
+                const auto it = this->ongoingConnectAttempts.find(nodeId);
+                if (it == this->ongoingConnectAttempts.end()) {
+                    return;
+                }
+                if (it->second.connection.get() != connPtr) {
+                    return;
+                }
+                this->ongoingConnectAttempts.erase(it);
             });
         pendingConnection->on<pendingconnectionevents::Disconnected>(
-            [this, nodeId](bool /*gracefulLeave*/) {
-                std::scoped_lock lock(this->mMutex);
-                this->ongoingConnectAttempts.erase(nodeId);
+            [this, nodeId, pendingPtr = pendingConnection.get()](
+                bool /*gracefulLeave*/) {
+                // The attempt's WebrtcConnection must be closed HERE, not
+                // just dropped from the map: for an answerer that never
+                // received a handshake request, no handshaker links the
+                // pending connection's Disconnected to connection->close()
+                // (IncomingHandshaker wires that link only inside
+                // onHandshakeRequest), so a bare erase would remove the last
+                // owner stop() could still destroy and leave the connection
+                // orphaned (see the immortality note above — this was the
+                // 1-in-10 post-PASSED exit hang of the 22-node webrtc test).
+                // close() is idempotent, so the offerer path (whose
+                // OutgoingHandshaker closes the connection from this same
+                // event first) is unaffected. Identity-checked like the
+                // handler above.
+                std::shared_ptr<WebrtcConnection> connectionToClose;
+                {
+                    std::scoped_lock lock(this->mMutex);
+                    const auto it = this->ongoingConnectAttempts.find(nodeId);
+                    if (it == this->ongoingConnectAttempts.end() ||
+                        it->second.managedConnection.get() != pendingPtr) {
+                        return;
+                    }
+                    connectionToClose = it->second.connection;
+                    this->ongoingConnectAttempts.erase(it);
+                }
+                // Outside mMutex (locking policy: no call-outs under the
+                // connector mutex — the close emits Disconnected and defers
+                // the rtc teardown to the worker pool).
+                if (connectionToClose) {
+                    connectionToClose->close(false);
+                }
             });
         pendingConnection->on<pendingconnectionevents::Connected>(
-            [this, nodeId](
+            [this, nodeId, pendingPtr = pendingConnection.get()](
                 const PeerDescriptor& /*peerDescriptor*/,
                 const std::shared_ptr<Connection>& /*connection*/) {
                 std::scoped_lock lock(this->mMutex);
-                this->ongoingConnectAttempts.erase(nodeId);
+                const auto it = this->ongoingConnectAttempts.find(nodeId);
+                if (it == this->ongoingConnectAttempts.end() ||
+                    it->second.managedConnection.get() != pendingPtr) {
+                    return;
+                }
+                this->ongoingConnectAttempts.erase(it);
             });
 
         connection->webrtcEvents().on<webrtcconnectionevents::LocalCandidate>(

packages/streamr-dht/modules/connection/websocket/WebsocketConnection.cppm

@@ -7,12 +7,15 @@ module;
 #include <functional>
 #include <rtc/rtc.hpp>
 
+#include <folly/synchronization/Baton.h>
+
 #include <string>
 
 export module streamr.dht.WebsocketConnection;
 
 import streamr.logger.SLogger;
 import streamr.utils.EnableSharedFromThis;
+import streamr.utils.SharedExecutors;
 import streamr.dht.Connection;
 
 // Hoisted from the former header (file scope, NOT exported);
@@ -43,6 +46,33 @@ protected:
     std::atomic<bool> mConnectedEmitted{false}; // NOLINT
     std::recursive_mutex mMutex; // NOLINT
 
+private:
+    // All libdatachannel callbacks are dispatched through this per-connection
+    // serial view of the shared worker pool instead of running inline. Every
+    // websocket in the process shares ONE libdatachannel poll thread; running
+    // the emit chains (handshake, RPC parse/dispatch) inline there starved
+    // concurrent websocket upgrades so badly that a 12-node start storm
+    // pushed connect latency past the 1 s connectivity-check timeout
+    // (measured: the poll thread was >95% busy in Data emit chains while
+    // accepts sat waiting). The serial executor preserves the per-connection
+    // event order the emitter contract relies on: Connected before any Data,
+    // Data in arrival order, Disconnected last (the same per-association
+    // ordering idiom as the Simulator's delivery executors).
+    streamr::utils::SharedSerialExecutor mDispatchExecutor{
+        streamr::utils::SharedExecutors::worker()};
+
+    // Enqueue a callback body; the task keeps the connection alive until it
+    // has run. Callers are live rtc callbacks (rtc's callback mutex
+    // guarantees the object cannot be mid-destruction there — the
+    // destructor's resetCallbacks() blocks on in-flight callbacks) or
+    // methods invoked through an owning shared_ptr (setSocket's
+    // already-open path), so sharedFromThis is safe.
+    void dispatch(std::function<void()> task) {
+        auto self = this->sharedFromThis<WebsocketConnection>();
+        mDispatchExecutor.add([self, task = std::move(task)]() { task(); });
+    }
+
+protected:
     // Only allow subclasses to be created
     explicit WebsocketConnection(ConnectionType type)
         : Connection(type),
@@ -157,23 +187,32 @@ protected:
 
         SLogger::trace("setSocket() after move " + getConnectionTypeString());
 
-        // Set socket callbacks. The message callback is deliberately NOT
-        // attached here — see startReceiving().
+        // Set socket callbacks. Each is a thin wrapper that only enqueues
+        // the real body onto the per-connection dispatch executor — the rtc
+        // callback threads (the shared poll thread, the per-server accept
+        // thread) must never run the emit chains (see mDispatchExecutor).
+        // The message callback is deliberately NOT attached here — see
+        // startReceiving().
 
-        mSocket->onError(this->onError);
-        mSocket->onClosed(this->onClosed);
-        mSocket->onOpen(this->onOpen);
+        mSocket->onError([this](std::string error) {
+            this->dispatch(
+                [this, error = std::move(error)]() { this->onError(error); });
+        });
+        mSocket->onClosed(
+            [this]() { this->dispatch([this]() { this->onClosed(); }); });
+        mSocket->onOpen(
+            [this]() { this->dispatch([this]() { this->onOpen(); }); });
 
         // rtc does not retro-fire the open callback: if the websocket
         // handshake completed on the processor thread before the callback
         // above was attached, onOpen would never run and Connected would
-        // never be emitted. Emit it here in that case (mConnectedEmitted
+        // never be emitted. Enqueue it here in that case (mConnectedEmitted
         // keeps the two paths idempotent).
         if (mSocket->readyState() == rtc::WebSocket::State::Open) {
             SLogger::trace(
                 "setSocket() socket already open, emitting Connected " +
                 getConnectionTypeString());
-            this->onOpen();
+            this->dispatch([this]() { this->onOpen(); });
         }
 
         SLogger::trace("setSocket() end " + getConnectionTypeString());
@@ -203,7 +242,17 @@ public:
         SLogger::trace("startReceiving() " + getConnectionTypeString());
         std::scoped_lock lock(mMutex);
         if (!mDestroyed && mSocket) {
-            mSocket->onMessage(this->onMessage, this->onStringMessage);
+            // Thin wrapper (see setSocket): the poll thread only copies the
+            // frame into a task; the emit<Data> chain runs on the dispatch
+            // executor. The attach-time flush of queued frames enqueues
+            // them here in order, ahead of any later-arriving frame.
+            mSocket->onMessage(
+                [this](rtc::binary data) {
+                    this->dispatch([this, data = std::move(data)]() mutable {
+                        this->onMessage(std::move(data));
+                    });
+                },
+                this->onStringMessage);
         }
     }
 
@@ -287,6 +336,19 @@ public:
             socket->close();
         }
     }
+
+    // Barrier: returns once every dispatch task enqueued so far has run.
+    // resetCallbacks() only stops NEW rtc callbacks; already-enqueued
+    // bodies still run afterwards, so an owner whose listeners capture it
+    // by raw pointer must drain before destroying itself (WebsocketServer
+    // does, for half-ready connections). MUST NOT be called from a task
+    // running on this connection's own dispatch executor — that would
+    // deadlock the serial queue.
+    void drainDispatchQueue() {
+        folly::Baton<> baton;
+        mDispatchExecutor.add([&baton]() { baton.post(); });
+        baton.wait();
+    }
 };
 
 } // namespace streamr::dht::connection::websocket

packages/streamr-dht/modules/connection/websocket/WebsocketServer.cppm

@@ -112,19 +112,37 @@ public:
 
     void stop() {
         SLogger::info("stop() start");
-        if (mStopped) {
+        if (mStopped.exchange(true)) {
             SLogger::info("stop() already stopped, returning");
             return;
         }
-        mStopped = true;
-        removeAllListeners();
-        SLogger::info(
-            "stop() removeAllListeners() end, calling mServer->stop()");
+        // Quiesce the accept thread first: mServer->stop() joins it, and
+        // handleIncomingClient is gated on mStopped, so no new half-ready
+        // connection can be inserted after this.
         if (mServer) {
             mServer->stop();
             mServer = nullptr;
         }
-        SLogger::info("stop() mServer->stop() end");
+        // Tear down the half-ready connections. Their listeners capture
+        // this server by raw pointer and run on the connections' dispatch
+        // executors, so each queue must be drained before the caller may
+        // destroy this object (removeAllListeners() first makes queued
+        // not-yet-run emits no-ops).
+        std::unordered_map<
+            std::string,
+            std::shared_ptr<WebsocketServerConnection>>
+            halfReady;
+        {
+            std::scoped_lock lock(mHalfReadyConnectionsMutex);
+            halfReady.swap(mHalfReadyConnections);
+        }
+        for (auto& [id, connection] : halfReady) {
+            connection->removeAllListeners();
+            connection->destroy();
+            connection->drainDispatchQueue();
+        }
+        removeAllListeners();
+        SLogger::info("stop() end");
     }
 
     void updateCertificate(const std::string& cert, const std::string& key) {
@@ -142,6 +160,10 @@ private:
 
     void handleIncomingClient(std::shared_ptr<rtc::WebSocket> ws) {
         std::scoped_lock lock(mHalfReadyConnectionsMutex);
+        if (mStopped) {
+            // Racing stop(): dropping ws here closes the socket.
+            return;
+        }
         auto websocketServerConnection =
             WebsocketServerConnection::newInstance();
         auto id = Uuid::v4();
@@ -151,7 +173,12 @@ private:
             SLogger::info(
                 "Half-ready WebSocketServerConnection emitted Connected, emitting it as Connected");
             std::scoped_lock lock(mHalfReadyConnectionsMutex);
-            auto readyConnection = mHalfReadyConnections.at(id);
+            const auto entry = mHalfReadyConnections.find(id);
+            if (entry == mHalfReadyConnections.end()) {
+                // stop() swapped the map out while this task was queued.
+                return;
+            }
+            auto readyConnection = entry->second;
 
             readyConnection->removeAllListeners();
 

packages/streamr-dht/modules/connection/websocket/WebsocketServerConnection.cppm

@@ -37,6 +37,15 @@ private:
                 "WebsocketServerConnection() The half-ready socket is fully connected");
             mResourceURL = Url{mSocket->path().value()};
             mRemoteAddress = mSocket->remoteAddress().value();
+            // libdatachannel formats this as "<host>:<service>"; the
+            // consumers (TS-parity remoteAddress, the connectivity
+            // probe URL, ipv4ToNumber) need the bare host. The service
+            // is always the suffix after the LAST colon, so this also
+            // handles unbracketed IPv6 hosts.
+            const auto portSep = mRemoteAddress.rfind(':');
+            if (portSep != std::string::npos) {
+                mRemoteAddress = mRemoteAddress.substr(0, portSep);
+            }
             this->off<connectionevents::Connected>(
                 this->mHalfReadyConnectedHandlerToken);
         });

packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm

@@ -175,6 +175,18 @@ public:
                                    const std::shared_ptr<
                                        WebsocketServerConnection>&
                                        serverSocket) {
+                if (this->abortController.getSignal().aborted) {
+                    // destroy() has run (or is running): drop the late
+                    // connection instead of touching the cleared maps.
+                    try {
+                        serverSocket->close(false);
+                    } catch (...) {
+                        SLogger::debug(
+                            "Error closing incoming Websocket connection"
+                            " after abort");
+                    }
+                    return;
+                }
                 const auto resourceUrl = serverSocket->getResourceURL();
                 const auto action = getActionFromUrl(resourceUrl);
                 SLogger::trace(