Fix Layer0 e2e teardown SIGSEGVs: drain RPC scopes before their targets die, roll back started on failed start()

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");
     };
 

packages/streamr-dht/modules/dht/DhtNode.cppm

@@ -175,12 +175,17 @@ private:
 
     DhtNodeOptions options;
     LocalDataStore localDataStore;
-    std::unique_ptr<RoutingRpcCommunicator> 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<streamr::dht::connection::ConnectionManager>
         ownedConnectionManager;
+    std::unique_ptr<RoutingRpcCommunicator> rpcCommunicator;
+    Transport* transportPtr = nullptr;
     ConnectionsView* connectionsView = nullptr;
     ConnectionLocker* connectionLocker = nullptr;
     std::shared_ptr<ConnectionLocker>
@@ -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;
     }
 

packages/streamr-dht/modules/dht/recursive-operation/RecursiveOperationManager.cppm

@@ -116,6 +116,32 @@ private:
         ongoingSessions;
     bool stopped = false;
     std::unique_ptr<RecursiveOperationRpcLocal> 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<std::shared_ptr<ListeningRpcCommunicator>>
+        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<ListeningRpcCommunicator>& 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<ListeningRpcCommunicator>(
+                    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<std::shared_ptr<ListeningRpcCommunicator>> 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();
     }
 };
 

packages/streamr-dht/modules/transport/ListeningRpcCommunicator.cppm

@@ -78,6 +78,13 @@ public:
 
     void destroy() {
         transport.off<transportevents::Message>(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<transportevents::Disconnected>(
+            this->onDisconnectedHandlerToken);
     }
     using RpcCommunicator::registerRpcMethod;
     using RpcCommunicator::registerRpcNotification;

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) {

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
      *

packages/streamr-proto-rpc/modules/RpcCommunicatorClientApi.cppm

@@ -8,6 +8,7 @@ module;
 // translation unit; it cannot arrive through an imported BMI.
 #include <coroutine> // IWYU pragma: keep
 
+#include <atomic>
 #include <exception>
 #include <map>
 #include <mutex>
@@ -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<bool> 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;

packages/streamr-proto-rpc/modules/RpcCommunicatorServerApi.cppm

@@ -19,6 +19,7 @@ module;
 // imported BMI.
 #include <coroutine> // IWYU pragma: keep
 
+#include <atomic>
 #include <exception>
 #include <functional>
 #include <optional>
@@ -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<bool> 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;