Fix the full-node teardown hang: cancel and suspend the joinScope drain

packages/streamr-dht/modules/connection/ConnectionLocker.cppm

@@ -8,6 +8,8 @@ module;
 
 #include <string>
 
+#include <folly/coro/Task.h>
+
 export module streamr.dht.ConnectionLocker;
 
 import streamr.dht.protos;
@@ -26,9 +28,16 @@ protected:
 public:
     virtual ~ConnectionLocker() = default;
 
-    virtual void lockConnection(
+    // Coroutines, not blocking calls: locking awaits a lockRequest RPC
+    // round-trip. The former synchronous implementation blockingWait()ed
+    // for it — on the shared worker pool that is a self-deadlock once
+    // poolsize concurrent joins sit inside lockConnection at once (every
+    // thread waits for an RPC response only those same threads could
+    // process; seen wedging the full-node teardown on the 4-thread CI
+    // pool).
+    virtual folly::coro::Task<void> lockConnection(
         PeerDescriptor targetDescriptor, LockID lockId) = 0;
-    virtual void unlockConnection(
+    virtual folly::coro::Task<void> unlockConnection(
         PeerDescriptor targetDescriptor, LockID lockId) = 0;
     virtual void weakLockConnection(
         const DhtAddress& targetDescriptor, const LockID& lockId) = 0;

packages/streamr-dht/modules/connection/ConnectionManager.cppm

@@ -473,12 +473,12 @@ public:
         return this->locks.isRemoteLocked(nodeId);
     }
 
-    void lockConnection(
+    folly::coro::Task<void> lockConnection(
         PeerDescriptor targetDescriptor, LockID lockId) override {
         if (this->state == ConnectionManagerState::STOPPED ||
             Identifiers::areEqualPeerDescriptors(
                 targetDescriptor, this->getLocalPeerDescriptor())) {
-            return;
+            co_return;
         }
         const auto nodeId =
             Identifiers::getNodeIdFromPeerDescriptor(targetDescriptor);
@@ -490,8 +490,9 @@ public:
         this->locks.addLocalLocked(nodeId, lockId);
 
         try {
-            auto accepted = streamr::utils::blockingWait(
-                rpcRemote.lockRequest(std::move(lockId)));
+            // co_await, never blockingWait: this runs on shared worker
+            // pool threads (see ConnectionLocker::lockConnection).
+            auto accepted = co_await rpcRemote.lockRequest(std::move(lockId));
             if (accepted) {
                 SLogger::trace("LockRequest successful");
             } else {
@@ -504,12 +505,12 @@ public:
         }
     }
 
-    void unlockConnection(
+    folly::coro::Task<void> unlockConnection(
         PeerDescriptor targetDescriptor, LockID lockId) override {
         if (this->state == ConnectionManagerState::STOPPED ||
             Identifiers::areEqualPeerDescriptors(
                 targetDescriptor, this->getLocalPeerDescriptor())) {
-            return;
+            co_return;
         }
 
         const auto nodeId =
@@ -522,16 +523,15 @@ public:
             SLogger::debug("Acquired mutex lock in unlockConnection");
             if (!this->endpoints.contains(nodeId)) {
                 SLogger::debug("Node ID not found in endpoints");
-                return;
+                co_return;
             }
         }
 
         ConnectionLockRpcClient client{this->rpcCommunicator};
         ConnectionLockRpcRemote rpcRemote(
             this->getLocalPeerDescriptor(), targetDescriptor, client);
 
-        streamr::utils::blockingWait(
-            rpcRemote.unlockRequest(std::move(lockId)));
+        co_await rpcRemote.unlockRequest(std::move(lockId));
     }
 
     void weakLockConnection(

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

@@ -157,14 +157,22 @@ public:
     }
 
     void destroy() {
-        std::scoped_lock lock(this->mutex);
-        this->abortController.abort();
-
-        for (const auto& [_, handshaker] : this->connectingHandshakers) {
+        // Snapshot under the lock, close() OUTSIDE it: close(true) fans
+        // out into Disconnected handlers that re-enter connection and
+        // connector code — holding this->mutex across that call-out
+        // deadlocks ABBA against a thread that holds a connection's
+        // mutex (an rtc close callback, a pool worker in connect()) and
+        // needs this->mutex in the HandshakerStopped handler (sampled
+        // wedging the full-node teardown).
+        std::map<DhtAddress, std::shared_ptr<OutgoingHandshaker>> handshakers;
+        {
+            std::scoped_lock lock(this->mutex);
+            this->abortController.abort();
+            handshakers.swap(this->connectingHandshakers);
+        }
+        for (const auto& [_, handshaker] : handshakers) {
             handshaker->getPendingConnection()->close(true);
         }
-
-        this->connectingHandshakers.clear();
     }
 };
 

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

@@ -352,34 +352,41 @@ public:
 
     void destroy() {
         SLogger::trace("WebsocketServerConnector::destroy() called");
+        std::map<std::string, std::shared_ptr<IncomingHandshaker>> handshakers;
+        std::map<DhtAddress, std::shared_ptr<IPendingConnection>> requests;
+        std::unique_ptr<WebsocketServer> server;
         {
             std::scoped_lock lock(this->mMutex);
             this->abortController.abort();
-
-            SLogger::trace("Closing ongoing connect requests");
-            for (const auto& request : this->ongoingConnectRequests) {
-                if (request.second) {
-                    request.second->close(true);
-                }
+            handshakers.swap(this->connectingHandshakers);
+            requests.swap(this->ongoingConnectRequests);
+            server = std::move(this->websocketServer);
+        }
+        // Every call-out below runs OUTSIDE mMutex: close(true) fans out
+        // into Disconnected handlers that re-enter connection/connector
+        // code, and holding the lock across that deadlocks ABBA against a
+        // thread that holds a connection's mutex and needs mMutex (the
+        // shape sampled in WebsocketClientConnector::destroy(); same
+        // hazard here).
+        SLogger::trace("Closing ongoing connect requests");
+        for (const auto& request : requests) {
+            if (request.second) {
+                request.second->close(true);
             }
+        }
 
-            SLogger::trace("Closing connecting handshakers");
-            for (const auto& handshaker : this->connectingHandshakers) {
-                if (handshaker.second &&
-                    handshaker.second->getPendingConnection()) {
-                    handshaker.second->getPendingConnection()->close(true);
-                }
+        SLogger::trace("Closing connecting handshakers");
+        for (const auto& handshaker : handshakers) {
+            if (handshaker.second &&
+                handshaker.second->getPendingConnection()) {
+                handshaker.second->getPendingConnection()->close(true);
             }
+        }
 
-            SLogger::trace("Clearing maps");
-            this->connectingHandshakers.clear();
-            this->ongoingConnectRequests.clear();
-
-            SLogger::trace("Stopping websocket server");
-            if (this->websocketServer) {
-                this->websocketServer->stop();
-                this->websocketServer.reset();
-            }
+        SLogger::trace("Stopping websocket server");
+        if (server) {
+            server->stop();
+            server.reset();
         }
         // Drained OUTSIDE the mutex (the PeerManager::stop() pattern): a
         // detached requestConnection notification still in flight references

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

@@ -730,9 +730,15 @@ public:
         // recovery executor's destructor join). A KBucketEmpty still firing
         // mid-join is dropped by the scope's close() gate.
         this->recoveryScope.close();
-        this->transportPtr->off<transportevents::Message>(this->messageToken);
-        this->transportPtr->off<Connected>(this->connectedToken);
-        this->transportPtr->off<Disconnected>(this->disconnectedToken);
+        // The waiting variants: a Message handler mid-invocation on
+        // another thread may still schedule a detached server-RPC
+        // dispatch; plain off() returning early would let it slip past
+        // the drain below (seen as a getClosestPeers dispatch resuming
+        // into a destroyed DhtNode on the CI-sized worker pool).
+        this->transportPtr->offAndWait<transportevents::Message>(
+            this->messageToken);
+        this->transportPtr->offAndWait<Connected>(this->connectedToken);
+        this->transportPtr->offAndWait<Disconnected>(this->disconnectedToken);
         streamr::utils::blockingWait(this->storeManager->destroy());
         this->localDataStore.clear();
         if (this->peerManager) {
@@ -753,6 +759,18 @@ public:
             this->ownedConnectionManager->stop();
         }
         this->connectionLocker = nullptr;
+        // Cancel and join this node's detached RPC dispatch/request
+        // coroutines while every member they capture is still alive. The
+        // implicit drain in ~RoutingRpcCommunicator runs only during
+        // member destruction — after later-declared members are already
+        // gone — and a straggler server dispatch then resumes into freed
+        // memory (seen as a getClosestPeers dispatch crashing on the
+        // CI-sized worker pool). Placed LAST: the drain gates new scope
+        // adds off, and the steps above still send through this
+        // communicator (leave notices, final replications). No new server
+        // dispatch can arrive here — the transport listeners were removed
+        // with offAndWait() at the top of stop().
+        this->rpcCommunicator->drainAsyncTasks();
     }
 
     // --- Public API ------------------------------------------------------

packages/streamr-dht/modules/dht/discovery/DiscoverySession.cppm

@@ -253,8 +253,15 @@ public:
         for (size_t i = 0; i < this->options.parallelism; ++i) {
             workers.push_back(this->worker());
         }
+        // MERGE the caller's (ambient) cancellation token with the node's
+        // abort signal instead of replacing it: a stop()-time cancellation
+        // of the detached join must reach the workers' RPC awaits, or the
+        // scope drain waits a full session timeout for them (the full-node
+        // teardown hang).
         co_await streamr::utils::co_withCancellation(
-            this->options.abortSignal.getCancellationToken(),
+            streamr::utils::cancellationTokenMerge(
+                co_await streamr::utils::co_currentCancellationToken(),
+                this->options.abortSignal.getCancellationToken()),
             folly::coro::timeout(
                 folly::coro::collectAllRange(std::move(workers)), timeout));
     }

packages/streamr-dht/modules/dht/discovery/PeerDiscovery.cppm

@@ -175,6 +175,11 @@ private:
         std::vector<std::shared_ptr<DiscoverySession>> sessions,
         PeerDescriptor entryPointDescriptor,
         bool retry) {
+        // Snapshot the caller's cancellation token: a join cancelled by
+        // its owner's stop() must not schedule a rejoin (the detached
+        // rejoin would outlive the intent to stop).
+        const auto callerToken =
+            co_await streamr::utils::co_currentCancellationToken();
         try {
             for (const auto& session : sessions) {
                 {
@@ -186,7 +191,7 @@ private:
         } catch (const std::exception&) {
             SLogger::debug("DHT join timed out");
         }
-        if (!this->isStopped()) {
+        if (!this->isStopped() && !callerToken.isCancellationRequested()) {
             if (this->options.peerManager.getNeighborCount() == 0) {
                 if (retry) {
                     this->scheduleRejoin(entryPointDescriptor, rejoinDelay);
@@ -396,7 +401,7 @@ public:
             co_return;
         }
         if (this->options.connectionLocker != nullptr) {
-            this->options.connectionLocker->lockConnection(
+            co_await this->options.connectionLocker->lockConnection(
                 entryPointDescriptor, this->joinLockId());
         }
         this->options.peerManager.addContact(entryPointDescriptor);
@@ -412,7 +417,7 @@ public:
         co_await this->runSessions(
             std::move(sessions), entryPointDescriptor, retry);
         if (this->options.connectionLocker != nullptr) {
-            this->options.connectionLocker->unlockConnection(
+            co_await this->options.connectionLocker->unlockConnection(
                 entryPointDescriptor, this->joinLockId());
         }
     }

packages/streamr-dht/modules/dht/discovery/RingDiscoverySession.cppm

@@ -270,8 +270,13 @@ public:
         for (size_t i = 0; i < this->options.parallelism; ++i) {
             workers.push_back(this->worker());
         }
+        // MERGE the ambient cancellation with the abort signal (see
+        // DiscoverySession::findClosestNodes for why replacing it hangs
+        // the stop()-time scope drain).
         co_await streamr::utils::co_withCancellation(
-            this->options.abortSignal.getCancellationToken(),
+            streamr::utils::cancellationTokenMerge(
+                co_await streamr::utils::co_currentCancellationToken(),
+                this->options.abortSignal.getCancellationToken()),
             folly::coro::timeout(
                 folly::coro::collectAllRange(std::move(workers)), timeout));
     }

packages/streamr-dht/modules/dht/store/StoreManager.cppm

@@ -30,6 +30,7 @@ export module streamr.dht.StoreManager;
 import streamr.dht.protos;
 
 import streamr.utils.CoroutineHelper;
+import streamr.utils.GuardedAsyncScope;
 import streamr.utils.SharedExecutors;
 import streamr.utils.EnableSharedFromThis;
 import streamr.utils.ExecutorHelper;
@@ -88,6 +89,16 @@ private:
     // replication tasks pin `self` (sharedFromThis), exactly as before.
     streamr::utils::SharedSerialExecutor replicationExecutor{
         streamr::utils::SharedExecutors::worker()};
+    // The detached onContactAdded replications pin `self`, but their RPC
+    // remotes send through the owning DhtNode's communicator, which
+    // destroy() does NOT keep alive: a replication coroutine resuming
+    // after DhtNode::stop() dereferenced the freed communicator (SIGSEGV
+    // in RpcCommunicatorClientApi::notify<ReplicateDataRequest> resume,
+    // reproduced 13/16 under a CI-sized worker pool). Track them here and
+    // drain in destroy(), while the communicator and transport are still
+    // alive; cancel first so the drain does not wait on network progress.
+    streamr::utils::GuardedAsyncScope replicationScope;
+    folly::CancellationSource replicationCancellation;
     std::unique_ptr<StoreRpcLocal> rpcLocal;
 
     explicit StoreManager(StoreManagerOptions options)
@@ -113,19 +124,22 @@ private:
         }
     }
 
-    // Fire-and-forget the async replication onto the worker executor.
+    // Fire-and-forget the async replication onto the worker executor,
+    // tracked by replicationScope (see the member comment).
     void replicateDataToContact(
         const DataEntry& dataEntry, const PeerDescriptor& contact) {
         auto self = this->sharedFromThis<StoreManager>();
         DataEntry entry = dataEntry;
         PeerDescriptor target = contact;
-        streamr::utils::co_withExecutor(
-            &this->replicationExecutor,
-            folly::coro::co_invoke(
-                [self, entry, target]() -> folly::coro::Task<void> {
-                    co_await self->doReplicate(entry, target);
-                }))
-            .start();
+        this->replicationScope.add(
+            streamr::utils::co_withExecutor(
+                &this->replicationExecutor,
+                streamr::utils::co_withCancellation(
+                    this->replicationCancellation.getToken(),
+                    folly::coro::co_invoke(
+                        [self, entry, target]() -> folly::coro::Task<void> {
+                            co_await self->doReplicate(entry, target);
+                        }))));
     }
 
     void registerLocalRpcMethods() {
@@ -188,17 +202,21 @@ private:
                 auto self = this->sharedFromThis<StoreManager>();
                 DhtAddress key = dataKey;
                 PeerDescriptor node = newNode;
-                streamr::utils::co_withExecutor(
-                    &this->replicationExecutor,
-                    folly::coro::co_invoke(
-                        [self, key, node]() -> folly::coro::Task<void> {
-                            const auto dataEntries =
-                                self->options.localDataStore.values(key);
-                            for (const auto& dataEntry : dataEntries) {
-                                co_await self->doReplicate(dataEntry, node);
-                            }
-                        }))
-                    .start();
+                this->replicationScope.add(
+                    streamr::utils::co_withExecutor(
+                        &this->replicationExecutor,
+                        streamr::utils::co_withCancellation(
+                            this->replicationCancellation.getToken(),
+                            folly::coro::co_invoke(
+                                [self, key, node]() -> folly::coro::Task<void> {
+                                    const auto dataEntries =
+                                        self->options.localDataStore.values(
+                                            key);
+                                    for (const auto& dataEntry : dataEntries) {
+                                        co_await self->doReplicate(
+                                            dataEntry, node);
+                                    }
+                                }))));
             }
         } else if (!std::ranges::any_of(
                        storers, [this](const PeerDescriptor& peer) {
@@ -317,6 +335,13 @@ public:
     }
 
     folly::coro::Task<void> destroy() {
+        // Cancel and drain the detached onContactAdded replications while
+        // the communicator/transport they send through are still alive
+        // (DhtNode::stop() awaits this before tearing those down). The
+        // final leave-replication below is awaited directly and is not
+        // cancelled.
+        this->replicationCancellation.requestCancellation();
+        co_await this->replicationScope.closeAsync();
         co_await this->replicateDataToClosestNodes();
     }
 };