From cd4e4a49efccef7f97b92d43ffe1fef995d73a38 Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Sat, 11 Jul 2026 19:32:03 +0300 Subject: [PATCH 1/2] Fix Layer0 end-to-end teardown SIGSEGVs: drain RPC scopes before their targets die, roll back started on failed start() Two teardown crash mechanisms, both proven with runtime evidence (macOS crash reports from 2026-07-11 + instrumented log ordering + deterministic reproductions): 1. Drain-order use-after-free (crash report 16:33, fixture destruction in Layer0MixedConnectionTypesTest.TwoNonServerPeersJoinFirst): the client/server RPC APIs drain their CancellableAsyncScopes in their own destructors, which run AFTER the owning objects' members are destroyed. A straggler request task drained that late runs makeRpcRequest -> the outgoing callback -> sendFn -> transportPtr->send() against freed memory: DhtNode declared rpcCommunicator before ownedConnectionManager, so the ConnectionManager was destroyed before the drain, and the drained send crashed in WebsocketClientConnector::connect inserting into a freed map. Instrumentation showed the unsafe window opened on EVERY node teardown (350 ordering violations in 10 full-suite runs) with ~170 guarded post-stop straggler sends per run - the crash only needed one task still queued at fixture destruction. Fix: idempotent drainAsyncTasks() on RpcCommunicatorClientApi / RpcCommunicatorServerApi / RpcCommunicator; ~RoutingRpcCommunicator calls it so the drain runs while sendFn, ownServiceId, the owner and its transport are all still alive (ListeningRpcCommunicator inherits this); ~ConnectionManager drains its own lock-RPC communicator in its body while endpoints/endpointsMutex/state are alive (sendIfStopped responses bypass the stopped-state guard in send()); DhtNode declares ownedConnectionManager before rpcCommunicator so the stopped ConnectionManager outlives the drain - a send into a STOPPED ConnectionManager is a guarded no-op, into a destroyed one a use-after-free. 2. Failed start() poisons stop() (crash report 07:21, SEGV at 0x1a0 in Layer0Test::TearDown): start() set started=true before creating the owned ConnectionManager; if the websocket server failed to bind, transportPtr stayed null while started stayed true, and stop() then called off() through the null pointer (0x1a0 = emitter mutex offset). Reproduced deterministically by occupying the entry-point port. Fix: set started=true only at the end of start() (no suspension points inside; a failed start leaves the node safely stoppable and the ConnectionManager destructor stops whatever came up). Verification: blocked-port repro SIGSEGV(0x1a0) -> clean test failure; drain-order violations 350/10 runs -> 0/33 runs; full e2e suite 15/15 under load and 3x7/7 clean; ASan 18-run loop shows zero heap reports (3 aborts are the known ASan-internal unwind CHECK artifact, same rate as before the fix); streamr-dht unit 181/181, proto-rpc 24/24 + 4/4. Co-Authored-By: Claude Fable 5 --- .../modules/connection/ConnectionManager.cppm | 6 ++++ packages/streamr-dht/modules/dht/DhtNode.cppm | 20 ++++++++++--- .../transport/RoutingRpcCommunicator.cppm | 15 ++++++++++ .../modules/RpcCommunicator.cppm | 11 ++++++++ .../modules/RpcCommunicatorClientApi.cppm | 25 +++++++++++++---- .../modules/RpcCommunicatorServerApi.cppm | 28 +++++++++++++------ 6 files changed, 87 insertions(+), 18 deletions(-) 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/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..c4b4f431 100644 --- a/packages/streamr-proto-rpc/modules/RpcCommunicator.cppm +++ b/packages/streamr-proto-rpc/modules/RpcCommunicator.cppm @@ -85,6 +85,17 @@ 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 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..da9c80e6 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,36 @@ 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 } } + ~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..1e8ccc06 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,27 @@ 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 } } + ~RpcCommunicatorServerApi() { this->drainAsyncTasks(); } + RpcCommunicatorServerApi(const RpcCommunicatorServerApi&) = delete; RpcCommunicatorServerApi& operator=(const RpcCommunicatorServerApi&) = delete; From a875bcfb657187621d38c8d78335908a1718b050 Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Sat, 11 Jul 2026 22:24:40 +0300 Subject: [PATCH 2/2] Fix worker-pool deadlock in RecursiveOperationManager::sendResponse (linux-arm64 CI hang) The linux-arm64 leg of the PR run hung for the full 1200 s ctest timeout (twice) in DhtNodeExternalApiTest.ExternalStoreDataHappyPath. Root cause (pre-existing on main, reproduced deterministically on BOTH main and this branch by capping the shared worker pool at 1 thread; thread dumps show the identical deadlock on both): sendResponse() runs on shared worker-pool threads (it is called from RPC server handlers) and destroyed its stack-local per-session ListeningRpcCommunicator there. The destructor joins the communicator's async scopes, and that join parks the pool thread waiting for the communicator's own queued fire-and-forget send task - which needs a pool thread to run. Once every worker is parked in such a join (plausible on the 3-core CI runner's 4-thread pool under concurrent recursive-operation responses), the queued sends can never run and the pool deadlocks permanently; Simulator::stop()'s drain-wait then blocks the test forever. Fix: keep the per-session communicator alive past the handler frame in a retired-communicators list owned by the manager. Entries are destroyed opportunistically once their scopes report zero pending tasks (an empty-scope join completes without needing a pool thread, so it is safe anywhere; growth is bounded because every send task settles within sessionRpcTimeout) and finally in stop(), which runs on the owner's thread where a blocking join may safely wait for pool tasks. RpcCommunicator/ClientApi/ServerApi gain a pendingAsyncTaskCount() probe for this. ListeningRpcCommunicator::destroy() now also detaches the Disconnected listener - previously it dangled on the transport after destruction and could also add client errors to a retired communicator. Verification: with a 1-thread pool the test went from 2/2 permanent hangs (on both main and this branch) to 0 hangs / bounded 10 s failures, and with the CI-sized 4-thread pool from flaky to 15/15 passes; full local regression green (DhtNodeExternalApiTest 4/4, e2e suite 3x7/7, dht unit 181/181, proto-rpc 24/24 + 4/4). Co-Authored-By: Claude Fable 5 --- .../RecursiveOperationManager.cppm | 65 +++++++++++++++---- .../transport/ListeningRpcCommunicator.cppm | 7 ++ .../modules/RpcCommunicator.cppm | 10 +++ .../modules/RpcCommunicatorClientApi.cppm | 8 +++ .../modules/RpcCommunicatorServerApi.cppm | 6 ++ 5 files changed, 85 insertions(+), 11 deletions(-) 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-proto-rpc/modules/RpcCommunicator.cppm b/packages/streamr-proto-rpc/modules/RpcCommunicator.cppm index c4b4f431..8685727d 100644 --- a/packages/streamr-proto-rpc/modules/RpcCommunicator.cppm +++ b/packages/streamr-proto-rpc/modules/RpcCommunicator.cppm @@ -96,6 +96,16 @@ public: 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 da9c80e6..3dc2d142 100644 --- a/packages/streamr-proto-rpc/modules/RpcCommunicatorClientApi.cppm +++ b/packages/streamr-proto-rpc/modules/RpcCommunicatorClientApi.cppm @@ -169,6 +169,14 @@ public: } } + // 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; diff --git a/packages/streamr-proto-rpc/modules/RpcCommunicatorServerApi.cppm b/packages/streamr-proto-rpc/modules/RpcCommunicatorServerApi.cppm index 1e8ccc06..86edaeb5 100644 --- a/packages/streamr-proto-rpc/modules/RpcCommunicatorServerApi.cppm +++ b/packages/streamr-proto-rpc/modules/RpcCommunicatorServerApi.cppm @@ -230,6 +230,12 @@ public: } } + // 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;