From bafaa4c5f555cac7ee9a35441534f0de7d0961a3 Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Fri, 17 Jul 2026 10:41:02 +0300 Subject: [PATCH 1/3] Logger: early-out below the cheapest enabled level before marshalling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Filtered-out trace/debug calls still marshalled their metadata to JSON and scanned the environment per call (FollyLoggerImpl) — 20-90 ms per network message in Debug builds, dominating the connection delivery chains under the full-node tests. The constructor snapshots the minimum enabled level across the logger's own level and every LOG_LEVEL_ override (env vars cannot meaningfully change mid-process; messages that pass the bound are still re-checked per category as before), and log() returns before any marshalling when the message level is below it. Adopted from the accept-path investigation (session branch claude/nostalgic-bell-34f492, 2026-07-13). Co-Authored-By: Claude Fable 5 --- packages/streamr-logger/modules/Logger.cppm | 40 +++++++++++++++++++++ 1 file changed, 40 insertions(+) diff --git a/packages/streamr-logger/modules/Logger.cppm b/packages/streamr-logger/modules/Logger.cppm index ee6ae622..f673d5a1 100644 --- a/packages/streamr-logger/modules/Logger.cppm +++ b/packages/streamr-logger/modules/Logger.cppm @@ -4,6 +4,7 @@ // this file is now the source of truth. module; +#include #include #include #include @@ -11,6 +12,10 @@ module; #include #include +// The C symbol for the process environment: must be declared in the +// global module fragment (see FollyLoggerImpl). +extern "C" char** environ; // NOLINT + export module streamr.logger.Logger; import streamr.json.toJson; @@ -31,6 +36,15 @@ private: std::shared_ptr mLoggerImpl; StreamrLogLevel mLoggerLogLevel; nlohmann::json mContextBindings; + // Cheapest enabled level across this logger AND every + // LOG_LEVEL_ env override (those can only raise verbosity + // for a file category). log() early-outs below this bound BEFORE + // marshalling the metadata to JSON: the marshalling (plus the + // per-call environment scan in FollyLoggerImpl) ran on every + // filtered-out trace/debug call and dominated the per-message cost + // of the connection delivery chains (measured 20-90 ms per network + // message in Debug builds under the full-node tests). + int mMinEnabledLogLevelValue{0}; public: /** @@ -67,6 +81,25 @@ public: mContextBindings = ensureJsonObject(toJson(contextBindings), "contextBindings"); + // Snapshot the minimum enabled level once (see the member's + // comment). Env vars cannot meaningfully change mid-process, and + // FollyLoggerImpl re-reads them per message anyway for the + // messages that pass this bound. + int minEnabled = getStreamrLogLevelValue(mLoggerLogLevel); + for (char** env = environ; *env != nullptr; ++env) { // NOLINT + const std::string_view envVar{*env}; + if (envVar.starts_with(detail::envCategoryLogLevelName)) { + const auto eq = envVar.find('='); + if (eq != std::string_view::npos) { + const auto categoryLevel = getStreamrLogLevelByName( + envVar.substr(eq + 1), mLoggerLogLevel); + minEnabled = std::min( + minEnabled, getStreamrLogLevelValue(categoryLevel)); + } + } + } + mMinEnabledLogLevelValue = minEnabled; + if (!loggerImpl) { mLoggerImpl = std::make_shared(); } else { @@ -209,6 +242,13 @@ private: std::string_view msg, MetadataType metadata, const std::source_location& location) { + // Early-out below the cheapest enabled level (see + // mMinEnabledLogLevelValue): no category could emit this message, + // so skip the metadata marshalling and the LoggerImpl call. + if (getStreamrLogLevelValue(messageLogLevel) < + mMinEnabledLogLevelValue) { + return; + } // Merge the possible metadata with the context bindings auto metadataJson = ensureJsonObject(toJson(metadata), "metadata"); From 3e61c5a117c14c3aa78a915f364f8ad3a275049d Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Fri, 17 Jul 2026 10:41:22 +0300 Subject: [PATCH 2/3] Websocket/Webrtc connections: dispatch rtc callbacks off the rtc threads MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit libdatachannel funnels ALL websocket TCP I/O through ONE global poll thread; websocket upgrades and message delivery ran inline there, and data channels share the fixed-size rtc worker pool with ICE/DTLS processing. Running the emit chains (handshake, RPC parse/dispatch) inline starved every other connection in the process: measured the poll thread >95% busy in Data emit chains while 12 concurrent connectivity checks timed out at their 1 s limit — the accept-path degradation that kept the full-node tests at 8 nodes. - WebsocketConnection/WebrtcConnection own a per-connection SharedSerialExecutor; every rtc callback is a thin wrapper that only enqueues the real body. Per-connection FIFO preserves the emitter contract (Connected before Data, Data in order, Disconnected last) — the Simulator's per-association idiom. - Teardown rules the offload requires: drainDispatchQueue() barrier; WebsocketServer::stop() quiesces the accept thread, swaps out the half-ready map and drains each connection (their listeners capture the server raw); handleIncomingClient gates on mStopped; the half-ready Connected handler uses find() (stop() may have swapped the map); WebsocketServerConnector's server-Connected handler drops late connections once aborted. - WebsocketServerConnection::getRemoteAddress(): libdatachannel returns host:service — strip at the LAST colon. The unstripped value made every connectivity probe dial ws://ip:port:port, so node descriptors lacked websocket info and links silently fell back to WebRTC. - Full-node tests restored to TS-parity node counts (websocket 12, webrtc 22). Adopted from the accept-path investigation (session branch claude/nostalgic-bell-34f492, 2026-07-13) and revalidated on top of the teardown fixes: websocket 12-node run now 3 s (was: failed SetUp beyond 8 nodes); on a CI-sized 4-thread pool 10/12 runs green with zero hangs, the two failures being the known pre-existing Handshaker-interleave UAF (follow-up filed). Co-Authored-By: Claude Fable 5 --- .../connection/webrtc/WebrtcConnection.cppm | 26 ++++++- .../websocket/WebsocketConnection.cppm | 78 +++++++++++++++++-- .../connection/websocket/WebsocketServer.cppm | 41 ++++++++-- .../websocket/WebsocketServerConnection.cppm | 9 +++ .../websocket/WebsocketServerConnector.cppm | 12 +++ .../test/unit/WebrtcFullNodeNetworkTest.cpp | 8 +- .../unit/WebsocketFullNodeNetworkTest.cpp | 8 +- 7 files changed, 153 insertions(+), 29 deletions(-) diff --git a/packages/streamr-dht/modules/connection/webrtc/WebrtcConnection.cppm b/packages/streamr-dht/modules/connection/webrtc/WebrtcConnection.cppm index 468132d2..f4e093f9 100644 --- a/packages/streamr-dht/modules/connection/webrtc/WebrtcConnection.cppm +++ b/packages/streamr-dht/modules/connection/webrtc/WebrtcConnection.cppm @@ -83,6 +83,13 @@ private: streamr::eventemitter::EventEmitter 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(message)) { - self->emit(std::get(std::move(message))); + // Thin enqueue (see onOpen above): the rtc worker thread + // only copies the frame; the emit chain runs on the + // per-connection serial executor in arrival order. + self->mDispatchExecutor.add( + [self, + data = + std::get(std::move(message))]() mutable { + self->emit(std::move(data)); + }); } }); } diff --git a/packages/streamr-dht/modules/connection/websocket/WebsocketConnection.cppm b/packages/streamr-dht/modules/connection/websocket/WebsocketConnection.cppm index 7fe34830..ce630dd0 100644 --- a/packages/streamr-dht/modules/connection/websocket/WebsocketConnection.cppm +++ b/packages/streamr-dht/modules/connection/websocket/WebsocketConnection.cppm @@ -7,12 +7,15 @@ module; #include #include +#include + #include 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 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 task) { + auto self = this->sharedFromThis(); + 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 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 diff --git a/packages/streamr-dht/modules/connection/websocket/WebsocketServer.cppm b/packages/streamr-dht/modules/connection/websocket/WebsocketServer.cppm index 101de48a..c0e58840 100644 --- a/packages/streamr-dht/modules/connection/websocket/WebsocketServer.cppm +++ b/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> + 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 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(); diff --git a/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnection.cppm b/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnection.cppm index ad26455f..44c341d8 100644 --- a/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnection.cppm +++ b/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 ":"; 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( this->mHalfReadyConnectedHandlerToken); }); diff --git a/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm b/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm index 2fc3acf1..9f15e1df 100644 --- a/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm +++ b/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( diff --git a/packages/streamr-trackerless-network/test/unit/WebrtcFullNodeNetworkTest.cpp b/packages/streamr-trackerless-network/test/unit/WebrtcFullNodeNetworkTest.cpp index 17d84e56..55390e13 100644 --- a/packages/streamr-trackerless-network/test/unit/WebrtcFullNodeNetworkTest.cpp +++ b/packages/streamr-trackerless-network/test/unit/WebrtcFullNodeNetworkTest.cpp @@ -43,12 +43,8 @@ using streamr::utils::waitForCondition; namespace { -// TS runs this with NUM_OF_NODES = 22; the C++ entry point's -// websocket accept path degrades under a concurrent connectivity-check -// storm (measured: ~9 accepts in the first second, then ~1/s), so the -// late checks exceed the 1 s connect timeout beyond ~8 nodes. Scaled -// down until the accept path is offloaded — see the follow-up task. -constexpr size_t numOfNodes = 8; +// TS parity: 22 nodes start concurrently against one entry point. +constexpr size_t numOfNodes = 22; constexpr uint16_t entryPointPort = 14444; // Generous for the 3-core CI runners: the TS test tolerates up to its // 220 s jest budget for convergence. diff --git a/packages/streamr-trackerless-network/test/unit/WebsocketFullNodeNetworkTest.cpp b/packages/streamr-trackerless-network/test/unit/WebsocketFullNodeNetworkTest.cpp index eefe23ea..b1fa6201 100644 --- a/packages/streamr-trackerless-network/test/unit/WebsocketFullNodeNetworkTest.cpp +++ b/packages/streamr-trackerless-network/test/unit/WebsocketFullNodeNetworkTest.cpp @@ -44,12 +44,8 @@ using streamr::utils::waitForCondition; namespace { -// TS runs this with NUM_OF_NODES = 12; the C++ entry point's -// websocket accept path degrades under a concurrent connectivity-check -// storm (measured: ~9 accepts in the first second, then ~1/s), so the -// late checks exceed the 1 s connect timeout beyond ~8 nodes. Scaled -// down until the accept path is offloaded — see the follow-up task. -constexpr size_t numOfNodes = 8; +// TS parity: 12 nodes start concurrently against one entry point. +constexpr size_t numOfNodes = 12; constexpr uint16_t entryPointPort = 15555; // Generous for the 3-core CI runners: the TS test tolerates up to its // 220 s jest budget for convergence. From e83beb547534362c87a6164c466dce3bab7f1f15 Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Fri, 17 Jul 2026 13:54:03 +0300 Subject: [PATCH 3/3] WebrtcConnector: close orphaned attempts; identity-check the attempt-map erases 1-in-10 runs of the 22-node WebrtcFullNodeNetworkTest hung AFTER gtest printed PASSED: exit-time static destruction blocked in CPUThreadPoolExecutor::~CPUThreadPoolExecutor -> joinKeepAliveOnce on a KeepAlive held by a leaked WebrtcConnection's dispatch executor. A WebrtcConnection is immortal until doClose() runs (its rtc callbacks and signalling listeners capture a strong self), and two attempt-map paths abandoned one without ever closing it: - pendingConnection Disconnected erased the attempt (the last owner stop() could still destroy) without closing the connection; for an answerer that never received a handshake request, IncomingHandshaker has no pending->connection close link yet, so nobody else closed it either. Triggers observed in lifecycle-registry logs: the pending connection's 15 s connect timeout and endpoint close at teardown. - a stale Disconnected from an older connection to the same nodeId erased a freshly inserted attempt (insert and erase within the same millisecond in the logs), orphaning the new pair at birth. All three erase handlers now erase only the pair that fired (raw-pointer identity comparison, never dereferenced), and the pendDisconnected handler closes the paired connection outside mMutex. Verified with repeated isolated runs under CPU load: baseline 4/15 exit-hangs (every hung process held exactly one never-closed answerer connection; clean runs held zero) -> 0/15 with the fix, plus 0/5 on the de-instrumented build; streamr-dht unit 181/181 and integration 43/43. Co-Authored-By: Claude Fable 5 --- .../connection/webrtc/WebrtcConnector.cppm | 68 ++++++++++++++++--- 1 file changed, 60 insertions(+), 8 deletions(-) diff --git a/packages/streamr-dht/modules/connection/webrtc/WebrtcConnector.cppm b/packages/streamr-dht/modules/connection/webrtc/WebrtcConnector.cppm index 344200ac..d0c54983 100644 --- a/packages/streamr-dht/modules/connection/webrtc/WebrtcConnector.cppm +++ b/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( - [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( - [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 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( - [this, nodeId]( + [this, nodeId, pendingPtr = pendingConnection.get()]( const PeerDescriptor& /*peerDescriptor*/, const std::shared_ptr& /*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(