diff --git a/packages/streamr-dht/modules/connection/ConnectionManager.cppm b/packages/streamr-dht/modules/connection/ConnectionManager.cppm index 77cc4c9f..5ab9e0df 100644 --- a/packages/streamr-dht/modules/connection/ConnectionManager.cppm +++ b/packages/streamr-dht/modules/connection/ConnectionManager.cppm @@ -235,6 +235,12 @@ public: SLogger::debug("~ConnectionManager() start"); SLogger::trace("~ConnectionManager()"); this->stop(); + // Drain the lock-RPC communicator's straggler coroutines while the + // members they reach (endpoints, endpointsMutex, state) are still + // alive — rpcCommunicator is declared before them, so its own + // destructor drain would run after they are destroyed, and + // sendIfStopped responses bypass the stopped-state guard in send(). + this->rpcCommunicator.drainAsyncTasks(); SLogger::debug("~ConnectionManager() end"); }; diff --git a/packages/streamr-dht/modules/dht/DhtNode.cppm b/packages/streamr-dht/modules/dht/DhtNode.cppm index 1557240c..2d808686 100644 --- a/packages/streamr-dht/modules/dht/DhtNode.cppm +++ b/packages/streamr-dht/modules/dht/DhtNode.cppm @@ -175,12 +175,17 @@ private: DhtNodeOptions options; LocalDataStore localDataStore; - std::unique_ptr rpcCommunicator; - Transport* transportPtr = nullptr; // Set only on the owned-ConnectionManager path (options.transport null); - // stop() stops it last, like TS. + // stop() stops it last, like TS. Declared BEFORE rpcCommunicator so it + // is destroyed AFTER it: the communicator's destructor drains straggler + // request/response coroutines that send through transportPtr, and a + // send into a STOPPED ConnectionManager is a guarded no-op while a send + // into a destroyed one is a use-after-free (the Layer0 end-to-end + // teardown SIGSEGV, macOS crash report 2026-07-11 16:33). std::shared_ptr ownedConnectionManager; + std::unique_ptr rpcCommunicator; + Transport* transportPtr = nullptr; ConnectionsView* connectionsView = nullptr; ConnectionLocker* connectionLocker = nullptr; std::shared_ptr @@ -437,7 +442,6 @@ public: if (this->started || this->abortController.getSignal().aborted) { co_return; } - this->started = true; if (this->options.transport != nullptr) { this->transportPtr = this->options.transport; this->connectionsView = this->options.connectionsView; @@ -607,6 +611,14 @@ public: key, operation); }}); this->bindRpcLocalMethods(); + // Set only after everything above succeeded: if start() throws + // (e.g. the websocket server port is taken), transportPtr is still + // null — a `started` node in that state would make stop() call + // off() through the null pointer (SIGSEGV at a near-null address + // inside the emitter mutex). A node whose start failed has nothing + // to stop; the owned ConnectionManager's destructor stops whatever + // did come up. + this->started = true; co_return; } diff --git a/packages/streamr-dht/modules/dht/recursive-operation/RecursiveOperationManager.cppm b/packages/streamr-dht/modules/dht/recursive-operation/RecursiveOperationManager.cppm index 840637a6..700b9b0f 100644 --- a/packages/streamr-dht/modules/dht/recursive-operation/RecursiveOperationManager.cppm +++ b/packages/streamr-dht/modules/dht/recursive-operation/RecursiveOperationManager.cppm @@ -116,6 +116,32 @@ private: ongoingSessions; bool stopped = false; std::unique_ptr rpcLocal; + // sendResponse()'s per-session communicators, kept alive until their + // fire-and-forget send task settles. sendResponse runs on shared + // worker-pool threads (RPC server handlers); destroying the + // communicator there joins its async scope, and that join parks the + // pool thread waiting for the communicator's own queued send task, + // which needs a pool thread — with every worker parked in such a join + // the queued sends can never run and the pool deadlocks permanently + // (linux-arm64 CI: DhtNodeExternalApiTest.ExternalStoreDataHappyPath + // hung its 1200 s timeout; reproduced deterministically with a + // 1-thread pool). Retired communicators are destroyed only once their + // scopes are empty (an empty-scope join needs no pool thread) or in + // stop(), which runs on the owner's thread, never on the pool. + std::vector> + retiredSessionCommunicators; + + // Called under this->mutex. Frees every retired communicator whose + // send task has settled; the rest stay retired until the next call or + // stop(). Growth is bounded: each pending entry settles within + // sessionRpcTimeout. + void pruneRetiredSessionCommunicators() { + std::erase_if( + this->retiredSessionCommunicators, + [](const std::shared_ptr& communicator) { + return communicator->pendingAsyncTaskCount() == 0; + }); + } explicit RecursiveOperationManager(RecursiveOperationManagerOptions options) : options(std::move(options)) {} @@ -210,21 +236,30 @@ private: dataEntries, noCloserNodesFound); } else { - ListeningRpcCommunicator remoteCommunicator( - ServiceID{serviceId}, - this->options.sessionTransport, - RpcCommunicatorOptions{.rpcRequestTimeout = sessionRpcTimeout}); + auto remoteCommunicator = + std::make_shared( + ServiceID{serviceId}, + this->options.sessionTransport, + RpcCommunicatorOptions{ + .rpcRequestTimeout = sessionRpcTimeout}); RecursiveOperationSessionRpcRemote rpcRemote( this->options.localPeerDescriptor, targetPeerDescriptor, - RecursiveOperationSessionRpcClient(remoteCommunicator), + RecursiveOperationSessionRpcClient(*remoteCommunicator), sessionRpcTimeout); rpcRemote.sendResponse( routingPath, closestConnectedNodes, dataEntries, noCloserNodesFound); - remoteCommunicator.destroy(); + remoteCommunicator->destroy(); + // Do NOT destroy the communicator here: this runs on a worker + // pool thread and the destructor's scope join would park it + // (see retiredSessionCommunicators). + std::scoped_lock lock(this->mutex); + this->pruneRetiredSessionCommunicators(); + this->retiredSessionCommunicators.push_back( + std::move(remoteCommunicator)); } } @@ -389,12 +424,20 @@ public: } void stop() { - std::scoped_lock lock(this->mutex); - this->stopped = true; - for (auto& [sessionId, session] : this->ongoingSessions) { - session->stop(); + std::vector> retiredToDestroy; + { + std::scoped_lock lock(this->mutex); + this->stopped = true; + for (auto& [sessionId, session] : this->ongoingSessions) { + session->stop(); + } + this->ongoingSessions.clear(); + retiredToDestroy.swap(this->retiredSessionCommunicators); } - this->ongoingSessions.clear(); + // Destroyed outside the mutex on the stopper's thread (never a + // pool worker), where the scope joins may safely wait for the + // in-flight send tasks to settle on the pool. + retiredToDestroy.clear(); } }; diff --git a/packages/streamr-dht/modules/transport/ListeningRpcCommunicator.cppm b/packages/streamr-dht/modules/transport/ListeningRpcCommunicator.cppm index 875a21cd..d73acc8e 100644 --- a/packages/streamr-dht/modules/transport/ListeningRpcCommunicator.cppm +++ b/packages/streamr-dht/modules/transport/ListeningRpcCommunicator.cppm @@ -78,6 +78,13 @@ public: void destroy() { transport.off(this->onMessageHandlerToken); + // Also detach the Disconnected listener: leaving it registered + // dangles `this` on the transport after destruction, and a live + // listener could still add client errors while a retired + // communicator waits to be destroyed (see + // RecursiveOperationManager::retiredSessionCommunicators). + transport.off( + this->onDisconnectedHandlerToken); } using RpcCommunicator::registerRpcMethod; using RpcCommunicator::registerRpcNotification; diff --git a/packages/streamr-dht/modules/transport/RoutingRpcCommunicator.cppm b/packages/streamr-dht/modules/transport/RoutingRpcCommunicator.cppm index 2051a0cd..56a2754a 100644 --- a/packages/streamr-dht/modules/transport/RoutingRpcCommunicator.cppm +++ b/packages/streamr-dht/modules/transport/RoutingRpcCommunicator.cppm @@ -114,6 +114,21 @@ public: this->sendFn(message, sendOpts); }); } + + // Drain the in-flight client/server coroutines while this object's + // members are still alive: the outgoing callback above reads + // ownServiceId and calls sendFn, and the base subobjects' own + // destructor drains only run AFTER those members are destroyed — a + // straggler request drained that late runs against destroyed members + // and a possibly-destroyed transport (observed as the Layer0 + // end-to-end teardown SIGSEGV, macOS crash reports 2026-07-11). + ~RoutingRpcCommunicator() { this->drainAsyncTasks(); } + + RoutingRpcCommunicator(const RoutingRpcCommunicator&) = delete; + RoutingRpcCommunicator& operator=(const RoutingRpcCommunicator&) = delete; + RoutingRpcCommunicator(RoutingRpcCommunicator&&) = delete; + RoutingRpcCommunicator& operator=(RoutingRpcCommunicator&&) = delete; + void handleMessageFromPeer(const Message& message) { if (message.serviceid() == this->ownServiceId && message.body_case() == Message::BodyCase::kRpcMessage) { diff --git a/packages/streamr-proto-rpc/modules/RpcCommunicator.cppm b/packages/streamr-proto-rpc/modules/RpcCommunicator.cppm index 705e4cca..8685727d 100644 --- a/packages/streamr-proto-rpc/modules/RpcCommunicator.cppm +++ b/packages/streamr-proto-rpc/modules/RpcCommunicator.cppm @@ -85,6 +85,27 @@ public: this->onIncomingMessage(message, callContext); } + /** + * @brief Cancel and join all in-flight client/server coroutines. + * Idempotent (the client/server API destructors re-run it as a no-op). + * Owners must call this before tearing down anything their outgoing + * message callback reaches (see RoutingRpcCommunicator's destructor). + */ + void drainAsyncTasks() noexcept { + mRpcCommunicatorClientApi.drainAsyncTasks(); + mRpcCommunicatorServerApi.drainAsyncTasks(); + } + + /** + * @brief Number of client/server coroutines still owned by the scopes. + * When zero, drainAsyncTasks()/destruction complete without needing a + * pool thread, so the communicator may be destroyed from any thread. + */ + [[nodiscard]] std::size_t pendingAsyncTaskCount() const noexcept { + return mRpcCommunicatorClientApi.pendingTaskCount() + + mRpcCommunicatorServerApi.pendingTaskCount(); + } + /** * @brief Set a callback for sending outgoing messages to the network * diff --git a/packages/streamr-proto-rpc/modules/RpcCommunicatorClientApi.cppm b/packages/streamr-proto-rpc/modules/RpcCommunicatorClientApi.cppm index c7e993fa..3dc2d142 100644 --- a/packages/streamr-proto-rpc/modules/RpcCommunicatorClientApi.cppm +++ b/packages/streamr-proto-rpc/modules/RpcCommunicatorClientApi.cppm @@ -8,6 +8,7 @@ module; // translation unit; it cannot arrive through an imported BMI. #include // IWYU pragma: keep +#include #include #include #include @@ -140,24 +141,44 @@ private: // scale to hundreds of nodes in one process, see // streamr.utils.SharedExecutors). folly::coro::CancellableAsyncScope mScope; + // drainAsyncTasks() must run the cancel-and-join exactly once (folly + // AsyncScope forbids a second join). + std::atomic mDrained = false; public: explicit RpcCommunicatorClientApi( std::chrono::milliseconds rpcRequestTimeout) : mRpcRequestTimeout(rpcRequestTimeout) {} - // Drains in-flight request/notify coroutines while all members are - // still alive. Must run on a thread other than the shared pool worker - // currently executing one of this instance's tasks (it always does: - // the communicator is owned and destroyed by the transport/main + // Drains in-flight request/notify coroutines. Idempotent. Owners whose + // straggler tasks reach through members that die before this subobject + // (RoutingRpcCommunicator's sendFn, ConnectionManager's endpoints) must + // call this BEFORE those members are destroyed — the destructor drain + // below is only a backstop that runs after the owning object's members + // are already gone. Must run on a thread other than the shared pool + // worker currently executing one of this instance's tasks (it always + // does: the communicator is owned and destroyed by the transport/main // thread, never from inside its own request path). - ~RpcCommunicatorClientApi() { + void drainAsyncTasks() noexcept { + if (mDrained.exchange(true)) { + return; + } try { streamr::utils::blockingWait(mScope.cancelAndJoinAsync()); - } catch (...) { // NOLINT(bugprone-empty-catch) dtor must not throw + } catch (...) { // NOLINT(bugprone-empty-catch) must not throw } } + // Number of request/notify coroutines still owned by the scope. When + // zero, drainAsyncTasks()/the destructor complete without needing any + // pool thread — the probe owners use to decide whether a retired + // communicator can be destroyed from a worker thread. + [[nodiscard]] std::size_t pendingTaskCount() const noexcept { + return mScope.remaining(); + } + + ~RpcCommunicatorClientApi() { this->drainAsyncTasks(); } + RpcCommunicatorClientApi(const RpcCommunicatorClientApi&) = delete; RpcCommunicatorClientApi& operator=(const RpcCommunicatorClientApi&) = delete; diff --git a/packages/streamr-proto-rpc/modules/RpcCommunicatorServerApi.cppm b/packages/streamr-proto-rpc/modules/RpcCommunicatorServerApi.cppm index 9bbdd462..86edaeb5 100644 --- a/packages/streamr-proto-rpc/modules/RpcCommunicatorServerApi.cppm +++ b/packages/streamr-proto-rpc/modules/RpcCommunicatorServerApi.cppm @@ -19,6 +19,7 @@ module; // imported BMI. #include // IWYU pragma: keep +#include #include #include #include @@ -58,6 +59,9 @@ private: streamr::utils::SharedSerialExecutor mSerialExecutor{ streamr::utils::SharedExecutors::worker()}; folly::coro::CancellableAsyncScope mScope; + // drainAsyncTasks() must run the cancel-and-join exactly once (folly + // AsyncScope forbids a second join). + std::atomic mDrained = false; struct RpcResponseParams { RpcMessage request; @@ -207,19 +211,33 @@ private: public: RpcCommunicatorServerApi() = default; - // Drains any in-flight response coroutines before the registry and the - // registry are destroyed, so no detached coroutine outlives the state it - // uses. Must run on a thread other than the serial executor's current - // worker (it always - // does: the communicator is owned and destroyed by the transport/main - // thread, never from inside a request handler). - ~RpcCommunicatorServerApi() { + // Drains any in-flight response coroutines before the registry is + // destroyed, so no detached coroutine outlives the state it uses. + // Idempotent; owners whose straggler responses reach through members + // that die before this subobject call it early (see + // RoutingRpcCommunicator's destructor), the destructor drain is the + // backstop. Must run on a thread other than the serial executor's + // current worker (it always does: the communicator is owned and + // destroyed by the transport/main thread, never from inside a request + // handler). + void drainAsyncTasks() noexcept { + if (mDrained.exchange(true)) { + return; + } try { streamr::utils::blockingWait(mScope.cancelAndJoinAsync()); - } catch (...) { // NOLINT(bugprone-empty-catch) dtor must not throw + } catch (...) { // NOLINT(bugprone-empty-catch) must not throw } } + // Number of response coroutines still owned by the scope (see + // RpcCommunicatorClientApi::pendingTaskCount()). + [[nodiscard]] std::size_t pendingTaskCount() const noexcept { + return mScope.remaining(); + } + + ~RpcCommunicatorServerApi() { this->drainAsyncTasks(); } + RpcCommunicatorServerApi(const RpcCommunicatorServerApi&) = delete; RpcCommunicatorServerApi& operator=(const RpcCommunicatorServerApi&) = delete;