From ff49b3ac23ce56cda87cb2af5b7facebd31335e0 Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Mon, 13 Jul 2026 19:29:24 +0300 Subject: [PATCH 1/8] Fix the full-node teardown hang: cancel and suspend the joinScope drain WebrtcFullNodeNetworkTest.HappyPath hit the 1200 s ctest timeout twice on the ubuntu-latest leg of PR #90: after one concurrent start threw "Timed out" (waitForNetworkConnectivity, 10 s), the process went silent for ~20 minutes until ctest killed it. Runtime-evidence diagnosis (logs + a worker-pool-saturation reproduction) confirmed the mechanism: - Every node's start() had already detached a fire-and-forget joinDht into NetworkStack's GuardedAsyncScope. TearDown drives all nine stop() coroutines on ONE blockingWait thread; the first stop()'s joinScope.close() blocked that thread AND waited for its join to complete NATURALLY (nothing cancelled it first, violating the scope's own cancel-before-close contract - the DhtNode abort only happens later in the stop sequence and cannot be reordered past the ContentDeliveryManager teardown-ordering constraints). - A join only completes with shared-worker-pool capacity and network progress: every coroutine resumption, including folly::coro::timeout resumptions, needs a pool slot. The 8-node start storm starved the 4-thread CI pool, so the drain never returned. Locally, saturating SharedExecutors::worker() reproduced the identical silent hang; the queued joinDht started the same millisecond a pool thread freed. The fix removes teardown's dependence on join completion: - NetworkStack: a CancellationSource wraps the detached joinDht; stop() requests cancellation BEFORE draining, and drains via the new suspending GuardedAsyncScope::closeAsync() so concurrent sibling stops sharing the drive thread proceed meanwhile. - DiscoverySession/RingDiscoverySession: MERGE the ambient cancellation token with the node's abort signal instead of replacing it - co_withCancellation replaces the ambient token, which made stop()-time cancellation invisible inside the sessions' RPC awaits. - PeerDiscovery::runSessions: a join cancelled by its owner's stop() no longer schedules a detached rejoin. - CoroutineHelper: export co_current_cancellation_token and cancellation_token_merge shims (CPOs, same pattern as blockingWait). Verified with the instrumented reproduction: post-fix, all nine joinScope drains enter in the same millisecond, cancelled joins collapse in ~1.5 s, and the failed-SetUp teardown takes 2.5 s (was: wedged for the whole pool-starvation window, 19+ minutes on CI). Under a 20 s artificial pool saturation, teardown completes ~5 s after the first pool liveness instead of never. GuardedAsyncScopeTest adds the deadlock-shape regression test; the dht end-to-end join suite (7 tests) passes under heavy load with the cancellation plumbing. Co-Authored-By: Claude Fable 5 --- .../dht/discovery/DiscoverySession.cppm | 9 ++- .../modules/dht/discovery/PeerDiscovery.cppm | 7 +- .../dht/discovery/RingDiscoverySession.cppm | 7 +- .../modules/NetworkStack.cppm | 25 ++++-- packages/streamr-utils/CMakeLists.txt | 1 + .../modules/CoroutineHelper.cppm | 20 +++++ .../modules/GuardedAsyncScope.cppm | 17 ++++ .../test/unit/GuardedAsyncScopeTest.cpp | 77 +++++++++++++++++++ 8 files changed, 154 insertions(+), 9 deletions(-) create mode 100644 packages/streamr-utils/test/unit/GuardedAsyncScopeTest.cpp diff --git a/packages/streamr-dht/modules/dht/discovery/DiscoverySession.cppm b/packages/streamr-dht/modules/dht/discovery/DiscoverySession.cppm index 270e5be8..9ea23225 100644 --- a/packages/streamr-dht/modules/dht/discovery/DiscoverySession.cppm +++ b/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)); } diff --git a/packages/streamr-dht/modules/dht/discovery/PeerDiscovery.cppm b/packages/streamr-dht/modules/dht/discovery/PeerDiscovery.cppm index 57f23a1a..e10d51e6 100644 --- a/packages/streamr-dht/modules/dht/discovery/PeerDiscovery.cppm +++ b/packages/streamr-dht/modules/dht/discovery/PeerDiscovery.cppm @@ -175,6 +175,11 @@ private: std::vector> 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); diff --git a/packages/streamr-dht/modules/dht/discovery/RingDiscoverySession.cppm b/packages/streamr-dht/modules/dht/discovery/RingDiscoverySession.cppm index 9a3b9af2..02a474ae 100644 --- a/packages/streamr-dht/modules/dht/discovery/RingDiscoverySession.cppm +++ b/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)); } diff --git a/packages/streamr-trackerless-network/modules/NetworkStack.cppm b/packages/streamr-trackerless-network/modules/NetworkStack.cppm index fe998cab..5615fbae 100644 --- a/packages/streamr-trackerless-network/modules/NetworkStack.cppm +++ b/packages/streamr-trackerless-network/modules/NetworkStack.cppm @@ -103,6 +103,12 @@ private: streamr::utils::SharedSerialExecutor joinExecutor{ streamr::utils::SharedExecutors::worker()}; GuardedAsyncScope joinScope; + // Cancels the fire-and-forget joinDht at stop(): the scope contract + // requires the tasks to be cancelled BEFORE the drain, otherwise + // stop() waits for the join to complete naturally — an unbounded + // wait on network progress (the CI full-node teardown hang: a + // starved shared worker pool never let the join finish). + folly::CancellationSource joinCancellation; folly::coro::Task ensureConnectedToControlLayer() { // TS TODO preserved: could wrap joinDht with pOnce and call it @@ -113,11 +119,13 @@ private: this->joinScope.add( streamr::utils::co_withExecutor( &this->joinExecutor, - folly::coro::co_invoke( - [this]() -> folly::coro::Task { - co_await this->controlLayerNode->joinDht( - this->options.layer0.entryPoints); - }))); + streamr::utils::co_withCancellation( + this->joinCancellation.getToken(), + folly::coro::co_invoke( + [this]() -> folly::coro::Task { + co_await this->controlLayerNode->joinDht( + this->options.layer0.entryPoints); + })))); } co_await this->controlLayerNode->waitForNetworkConnectivity(); } @@ -323,7 +331,12 @@ public: this->stopped = true; // Drain the fire-and-forget joins before the members die // (teardown ordering; see ContentDeliveryManager::destroy). - this->joinScope.close(); + // Cancel first — the drain must not wait for the join to make + // network progress — and suspend (not block) while draining, + // so concurrent sibling stops sharing this drive thread + // proceed meanwhile (the CI full-node teardown hang). + this->joinCancellation.requestCancellation(); + co_await this->joinScope.closeAsync(); co_await this->contentDeliveryManager->destroy(); // The info communicator listens on the transport — take it // down while the transport is still alive. diff --git a/packages/streamr-utils/CMakeLists.txt b/packages/streamr-utils/CMakeLists.txt index c9487917..faf1cd44 100644 --- a/packages/streamr-utils/CMakeLists.txt +++ b/packages/streamr-utils/CMakeLists.txt @@ -127,6 +127,7 @@ if(NOT IOS AND STREAMR_MODULES_SUPPORTED) add_executable(streamr-utils-test-unit test/unit/BrandedTest.cpp test/unit/AbortableTimersTest.cpp + test/unit/GuardedAsyncScopeTest.cpp test/unit/waitForEventTest.cpp test/unit/waitForConditionTest.cpp test/unit/runAndWaitForEventsTest.cpp diff --git a/packages/streamr-utils/modules/CoroutineHelper.cppm b/packages/streamr-utils/modules/CoroutineHelper.cppm index 54818fe9..a9bb0f57 100644 --- a/packages/streamr-utils/modules/CoroutineHelper.cppm +++ b/packages/streamr-utils/modules/CoroutineHelper.cppm @@ -26,6 +26,7 @@ module; #include #include #include +#include #include #include #include @@ -98,4 +99,23 @@ inline auto co_withCancellation( return folly::coro::co_withCancellation(std::forward(args)...); } +// The awaitable tag that yields the CURRENT (ambient) cancellation token +// inside a Task: co_await co_currentCancellationToken(). Exported so +// awaited subtrees can MERGE the caller's token with their own instead of +// replacing it (co_withCancellation replaces; a replaced token made +// stop()-time cancellation invisible inside DiscoverySession — the +// full-node teardown hang). +inline auto +co_currentCancellationToken() // NOLINT(readability-identifier-naming) + -> folly::coro::co_current_cancellation_token_t { + return folly::coro::co_current_cancellation_token; +} + +// folly::cancellation_token_merge is a CPO like the ones above. +template +inline auto cancellationTokenMerge(Args&&... args) + -> decltype(folly::cancellation_token_merge(std::forward(args)...)) { + return folly::cancellation_token_merge(std::forward(args)...); +} + } // namespace streamr::utils diff --git a/packages/streamr-utils/modules/GuardedAsyncScope.cppm b/packages/streamr-utils/modules/GuardedAsyncScope.cppm index d4518c49..4372173a 100644 --- a/packages/streamr-utils/modules/GuardedAsyncScope.cppm +++ b/packages/streamr-utils/modules/GuardedAsyncScope.cppm @@ -59,6 +59,23 @@ public: streamr::utils::blockingWait(this->scope.joinAsync()); } + // Coroutine variant of close() for stop() paths that share a drive + // thread (e.g. several stops under one collectAllRange): it SUSPENDS + // while the scope drains instead of blocking the thread, so sibling + // stops keep making progress. Same single-owning-stopper contract as + // close(); the drain-vs-destruction ordering is the caller's (the + // co_awaiting stop() completes before its owner is destroyed). + folly::coro::Task closeAsync() { + { + std::scoped_lock lock(this->mutex); + if (this->closed) { + co_return; + } + this->closed = true; + } + co_await this->scope.joinAsync(); + } + ~GuardedAsyncScope() { this->close(); } GuardedAsyncScope() = default; diff --git a/packages/streamr-utils/test/unit/GuardedAsyncScopeTest.cpp b/packages/streamr-utils/test/unit/GuardedAsyncScopeTest.cpp new file mode 100644 index 00000000..cc0ad6be --- /dev/null +++ b/packages/streamr-utils/test/unit/GuardedAsyncScopeTest.cpp @@ -0,0 +1,77 @@ +#include +#include +#include +#include + +#include // IWYU pragma: keep + +import streamr.utils.CoroutineHelper; +import streamr.utils.GuardedAsyncScope; +import streamr.utils.SharedExecutors; + +using streamr::utils::blockingWait; +using streamr::utils::GuardedAsyncScope; +using streamr::utils::SharedExecutors; + +namespace { + +auto onWorker(folly::coro::Task&& task) { + return streamr::utils::co_withExecutor( + &SharedExecutors::worker(), std::move(task)); +} + +} // namespace + +TEST(GuardedAsyncScopeTest, CloseAsyncDrainsAddedTasks) { + GuardedAsyncScope scope; + std::atomic ran{false}; + scope.add( + onWorker(folly::coro::co_invoke([&ran]() -> folly::coro::Task { + ran = true; + co_return; + }))); + blockingWait(scope.closeAsync()); + EXPECT_EQ(ran.load(), true); +} + +TEST(GuardedAsyncScopeTest, AddAfterCloseAsyncIsDropped) { + GuardedAsyncScope scope; + blockingWait(scope.closeAsync()); + std::atomic ran{false}; + scope.add( + onWorker(folly::coro::co_invoke([&ran]() -> folly::coro::Task { + ran = true; + co_return; + }))); + // close() after closeAsync() must stay a no-op (the destructor path). + scope.close(); + EXPECT_EQ(ran.load(), false); +} + +// The full-node teardown regression: several stop() coroutines share one +// blockingWait drive thread (collectAllRange), and the first scope's +// drain only finishes after a SIBLING stop has made progress. A +// thread-blocking close() deadlocks this shape; closeAsync() suspends, +// letting the sibling run, so the whole teardown completes. +TEST(GuardedAsyncScopeTest, CloseAsyncSuspendsSoSiblingStopsProgress) { + GuardedAsyncScope scope; + auto contract = folly::coro::makePromiseContract(); + scope.add(onWorker( + folly::coro::co_invoke( + [future = std::move(contract.second)]() mutable + -> folly::coro::Task { co_await std::move(future); }))); + + std::vector> stops; + stops.push_back( + folly::coro::co_invoke([&scope]() -> folly::coro::Task { + co_await scope.closeAsync(); + })); + stops.push_back( + folly::coro::co_invoke( + [promise = std::move( + contract.first)]() mutable -> folly::coro::Task { + promise.setValue(folly::Unit{}); + co_return; + })); + blockingWait(folly::coro::collectAllRange(std::move(stops))); +} From ceac17da77980f4a5c98c6b71bbc79a092a1e6be Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Fri, 17 Jul 2026 09:45:40 +0300 Subject: [PATCH 2/8] ContentDeliveryManager: cancel and suspend the joinScope drain in destroy() MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Same mechanism and fix as NetworkStack's joinScope in the parent commit, one layer down: destroy()'s blocking joinScope.close() drained un-cancelled startLayersAndJoinDht/handleEntryPointLeave tasks on the shared blockingWait drive thread — unbounded under a starved worker pool (sampled as ContentDeliveryManager.cppm:209 -> GuardedAsyncScope::close() wedging the WebrtcFullNodeNetworkTest teardown on a CI-sized 4-thread pool). A joinCancellation source now wraps all four detached joinScope tasks; destroy() requests cancellation first and drains via the suspending closeAsync(). Co-Authored-By: Claude Fable 5 --- .../modules/ContentDeliveryManager.cppm | 63 ++++++++++++------- 1 file changed, 40 insertions(+), 23 deletions(-) diff --git a/packages/streamr-trackerless-network/modules/ContentDeliveryManager.cppm b/packages/streamr-trackerless-network/modules/ContentDeliveryManager.cppm index 83d8f9a9..212fa7b9 100644 --- a/packages/streamr-trackerless-network/modules/ContentDeliveryManager.cppm +++ b/packages/streamr-trackerless-network/modules/ContentDeliveryManager.cppm @@ -158,6 +158,14 @@ private: streamr::utils::SharedSerialExecutor joinExecutor{ streamr::utils::SharedExecutors::worker()}; GuardedAsyncScope joinScope; + // Cancels the detached join/rejoin tasks at destroy(): the scope + // contract requires the tasks to be cancelled BEFORE the drain, + // otherwise destroy() waits for the joins to complete naturally — an + // unbounded wait on worker-pool/network progress. Same mechanism and + // fix as NetworkStack's joinCancellation; without it the CI + // full-node teardown wedged in ContentDeliveryManager::destroy() -> + // joinScope.close() one layer below the NetworkStack drain. + folly::CancellationSource joinCancellation; public: explicit ContentDeliveryManager(ContentDeliveryManagerOptions options) @@ -206,7 +214,13 @@ public: } } } - this->joinScope.close(); + // Cancel the detached join/rejoin tasks, then drain with the + // suspending closeAsync(): destroy() runs on a shared + // blockingWait drive thread during test teardown, and a blocking + // close() would both serialize sibling stops and wait unboundedly + // for un-cancelled joins (the CI full-node teardown hang). + this->joinCancellation.requestCancellation(); + co_await this->joinScope.closeAsync(); std::vector> parts; { std::scoped_lock lock(this->mutex); @@ -343,24 +357,27 @@ public: SLogger::debug( "Manual rejoin required for stream part " + streamPartId); - this->joinScope.add( - streamr::utils::co_withExecutor( - &this->joinExecutor, - streamPartReconnect->reconnect())); + this->joinScope.add(streamr::utils::co_withExecutor( + &this->joinExecutor, + streamr::utils::co_withCancellation( + this->joinCancellation.getToken(), + streamPartReconnect->reconnect()))); } }); node->on( [this, streamPartId, peerDescriptorStoreManager]() { - this->joinScope.add( - streamr::utils::co_withExecutor( - &this->joinExecutor, + this->joinScope.add(streamr::utils::co_withExecutor( + &this->joinExecutor, + streamr::utils::co_withCancellation( + this->joinCancellation.getToken(), this->handleEntryPointLeave( - streamPartId, peerDescriptorStoreManager))); + streamPartId, peerDescriptorStoreManager)))); }); // TS setImmediate(): detached bounded join. - this->joinScope.add( - streamr::utils::co_withExecutor( - &this->joinExecutor, + this->joinScope.add(streamr::utils::co_withExecutor( + &this->joinExecutor, + streamr::utils::co_withCancellation( + this->joinCancellation.getToken(), folly::coro::co_invoke( [this, streamPartId, peerDescriptorStoreManager]() -> folly::coro::Task { @@ -372,7 +389,7 @@ public: "Failed to join to stream part " + streamPartId + ": " + std::string(err.what())); } - }))); + })))); } folly::coro::Task setProxies( @@ -457,12 +474,11 @@ public: if (part->proxied) { continue; } - infos.push_back( - StreamPartitionInfo{ - .id = streamPartId, - .controlLayerNeighbors = - part->discoveryLayerNode->getNeighbors(), - .contentDeliveryLayerNeighbors = part->node->getInfos()}); + infos.push_back(StreamPartitionInfo{ + .id = streamPartId, + .controlLayerNeighbors = + part->discoveryLayerNode->getNeighbors(), + .contentDeliveryLayerNeighbors = part->node->getInfos()}); } return infos; } @@ -594,15 +610,16 @@ private: minNeighborCount) { auto networkSplitAvoidance = streamPart->networkSplitAvoidance; - this->joinScope.add( - streamr::utils::co_withExecutor( - &this->joinExecutor, + this->joinScope.add(streamr::utils::co_withExecutor( + &this->joinExecutor, + streamr::utils::co_withCancellation( + this->joinCancellation.getToken(), folly::coro::co_invoke( [networkSplitAvoidance]() -> folly::coro::Task { co_await networkSplitAvoidance ->avoidNetworkSplit(); - }))); + })))); } } } From 149b2fef82195904e18111fe9a6803521fd55842 Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Fri, 17 Jul 2026 09:46:00 +0300 Subject: [PATCH 3/8] StoreManager: scope and drain the detached onContactAdded replications MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The fire-and-forget replication coroutines were started with a bare .start(): they pin the StoreManager via sharedFromThis, but their RPC remotes send through the owning DhtNode's communicator, which nothing kept alive — a replication resuming after DhtNode::stop() dereferenced the freed communicator (SIGSEGV in RpcCommunicatorClientApi::notify resume, reproduced 13/16 on a CI-sized 4-thread pool once the teardown drain no longer serialized everything). Both launch sites now go through a cancellable GuardedAsyncScope that destroy() cancels and drains while the communicator and transport are still alive; the awaited final leave-replication is unchanged. Co-Authored-By: Claude Fable 5 --- .../modules/dht/store/StoreManager.cppm | 55 +++++++++++++------ 1 file changed, 38 insertions(+), 17 deletions(-) diff --git a/packages/streamr-dht/modules/dht/store/StoreManager.cppm b/packages/streamr-dht/modules/dht/store/StoreManager.cppm index 856c3b6c..a9a6803f 100644 --- a/packages/streamr-dht/modules/dht/store/StoreManager.cppm +++ b/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 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 rpcLocal; explicit StoreManager(StoreManagerOptions options) @@ -113,19 +124,21 @@ 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(); DataEntry entry = dataEntry; PeerDescriptor target = contact; - streamr::utils::co_withExecutor( + this->replicationScope.add(streamr::utils::co_withExecutor( &this->replicationExecutor, - folly::coro::co_invoke( - [self, entry, target]() -> folly::coro::Task { - co_await self->doReplicate(entry, target); - })) - .start(); + streamr::utils::co_withCancellation( + this->replicationCancellation.getToken(), + folly::coro::co_invoke( + [self, entry, target]() -> folly::coro::Task { + co_await self->doReplicate(entry, target); + })))); } void registerLocalRpcMethods() { @@ -188,17 +201,18 @@ private: auto self = this->sharedFromThis(); DhtAddress key = dataKey; PeerDescriptor node = newNode; - streamr::utils::co_withExecutor( + this->replicationScope.add(streamr::utils::co_withExecutor( &this->replicationExecutor, - folly::coro::co_invoke( - [self, key, node]() -> folly::coro::Task { - const auto dataEntries = - self->options.localDataStore.values(key); - for (const auto& dataEntry : dataEntries) { - co_await self->doReplicate(dataEntry, node); - } - })) - .start(); + streamr::utils::co_withCancellation( + this->replicationCancellation.getToken(), + folly::coro::co_invoke( + [self, key, node]() -> folly::coro::Task { + 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 +331,13 @@ public: } folly::coro::Task 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(); } }; From 520ed7f24e51c32358b3e939349ef84504b02a69 Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Fri, 17 Jul 2026 09:46:00 +0300 Subject: [PATCH 4/8] ConnectionLocker: make lockConnection/unlockConnection coroutines MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both wrapped their RPC round-trip in a thread-blocking blockingWait(). On the shared worker pool that is a self-deadlock: once poolsize concurrent joins sit inside lockConnection at the same time, every pool thread waits for a lock-RPC response that only those same threads could process (sampled: all four StreamrWorkers of a CI-sized pool blocked in PeerDiscovery::joinThroughEntryPoint -> lockConnection -> blockingWait, test wedged). The interface and the ConnectionManager implementation now return Task; PeerDiscovery co_awaits them. ProxyClient's synchronous connection routine wraps the calls in blockingWait at its own call sites for now — it already blocks on its other RPCs, and its callers block at the C-API boundary; async-ifying that path is a follow-up. Co-Authored-By: Claude Fable 5 --- .../modules/connection/ConnectionLocker.cppm | 13 +- .../modules/connection/ConnectionManager.cppm | 127 +++++++++--------- .../modules/dht/discovery/PeerDiscovery.cppm | 46 +++---- .../integration/ConnectionLockingTest.cpp | 27 ++-- .../modules/logic/proxy/ProxyClient.cppm | 120 +++++++++-------- .../test/utils/TestUtils.cppm | 17 ++- 6 files changed, 177 insertions(+), 173 deletions(-) diff --git a/packages/streamr-dht/modules/connection/ConnectionLocker.cppm b/packages/streamr-dht/modules/connection/ConnectionLocker.cppm index dd2a5c4c..4d297301 100644 --- a/packages/streamr-dht/modules/connection/ConnectionLocker.cppm +++ b/packages/streamr-dht/modules/connection/ConnectionLocker.cppm @@ -8,6 +8,8 @@ module; #include +#include + 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 lockConnection( PeerDescriptor targetDescriptor, LockID lockId) = 0; - virtual void unlockConnection( + virtual folly::coro::Task unlockConnection( PeerDescriptor targetDescriptor, LockID lockId) = 0; virtual void weakLockConnection( const DhtAddress& targetDescriptor, const LockID& lockId) = 0; diff --git a/packages/streamr-dht/modules/connection/ConnectionManager.cppm b/packages/streamr-dht/modules/connection/ConnectionManager.cppm index 5ab9e0df..33aae94b 100644 --- a/packages/streamr-dht/modules/connection/ConnectionManager.cppm +++ b/packages/streamr-dht/modules/connection/ConnectionManager.cppm @@ -166,43 +166,41 @@ public: this->send(message, sendOptions); }, RpcCommunicatorOptions{.rpcRequestTimeout = 10s}), // NOLINT - connectionLockRpcLocal( - ConnectionLockRpcLocalOptions{ - .addRemoteLocked = - [this](const DhtAddress& id, const LockID& lockId) { - this->locks.addRemoteLocked(id, lockId); - }, - .removeRemoteLocked = - [this](const DhtAddress& id, const LockID& lockId) { - this->locks.removeRemoteLocked(id, lockId); - }, - .closeConnection = - [this]( - const PeerDescriptor& peerDescriptor, - bool gracefulLeave, - const std::optional& reason) { + connectionLockRpcLocal(ConnectionLockRpcLocalOptions{ + .addRemoteLocked = + [this](const DhtAddress& id, const LockID& lockId) { + this->locks.addRemoteLocked(id, lockId); + }, + .removeRemoteLocked = + [this](const DhtAddress& id, const LockID& lockId) { + this->locks.removeRemoteLocked(id, lockId); + }, + .closeConnection = + [this]( + const PeerDescriptor& peerDescriptor, + bool gracefulLeave, + const std::optional& reason) { + SLogger::debug("closeConnection() callback of RpcLocal"); + this->closeConnection( + peerDescriptor, gracefulLeave, reason); + }, + .getLocalPeerDescriptor = + [this]() { return this->getLocalPeerDescriptor(); }, + .setPrivate = + [this](const DhtAddress& id, bool isPrivate) { + if (!this->options.allowIncomingPrivateConnections) { SLogger::debug( - "closeConnection() callback of RpcLocal"); - this->closeConnection( - peerDescriptor, gracefulLeave, reason); - }, - .getLocalPeerDescriptor = - [this]() { return this->getLocalPeerDescriptor(); }, - .setPrivate = - [this](const DhtAddress& id, bool isPrivate) { - if (!this->options.allowIncomingPrivateConnections) { - SLogger::debug( - "node " + id + - " attempted to set a connection as private," - " but it is not allowed"); - return; - } - if (isPrivate) { - this->locks.addPrivate(id); - } else { - this->locks.removePrivate(id); - } - }}) { + "node " + id + + " attempted to set a connection as private," + " but it is not allowed"); + return; + } + if (isPrivate) { + this->locks.addPrivate(id); + } else { + this->locks.removePrivate(id); + } + }}) { SLogger::debug("ConnectionManager constructor start"); SLogger::info("ConnectionManager constructor"); this->connectorFacade = this->options.createConnectorFacade(); @@ -473,12 +471,12 @@ public: return this->locks.isRemoteLocked(nodeId); } - void lockConnection( + folly::coro::Task 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 +488,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 +503,12 @@ public: } } - void unlockConnection( + folly::coro::Task unlockConnection( PeerDescriptor targetDescriptor, LockID lockId) override { if (this->state == ConnectionManagerState::STOPPED || Identifiers::areEqualPeerDescriptors( targetDescriptor, this->getLocalPeerDescriptor())) { - return; + co_return; } const auto nodeId = @@ -522,7 +521,7 @@ public: SLogger::debug("Acquired mutex lock in unlockConnection"); if (!this->endpoints.contains(nodeId)) { SLogger::debug("Node ID not found in endpoints"); - return; + co_return; } } @@ -530,8 +529,7 @@ public: ConnectionLockRpcRemote rpcRemote( this->getLocalPeerDescriptor(), targetDescriptor, client); - streamr::utils::blockingWait( - rpcRemote.unlockRequest(std::move(lockId))); + co_await rpcRemote.unlockRequest(std::move(lockId)); } void weakLockConnection( @@ -746,27 +744,24 @@ private: if (endpoint->isConnected()) { try { SLogger::debug("gracefullyDisconnect() calling blockingWait()"); - streamr::utils::blockingWait( - folly::coro::co_invoke( - [this, - endpoint, - targetDescriptor = std::move(targetDescriptor), - disconnectMode]() -> folly::coro::Task { - co_await folly::coro::collectAll( - waitForEvent( - endpoint.get(), 2000ms), // NOLINT - folly::coro::co_invoke( - [this, - endpoint, - targetDescriptor, - disconnectMode]() - -> folly::coro::Task { - co_return co_await this - ->doGracefullyDisconnectAsync( - targetDescriptor, - disconnectMode); - })); - })); + streamr::utils::blockingWait(folly::coro::co_invoke( + [this, + endpoint, + targetDescriptor = std::move(targetDescriptor), + disconnectMode]() -> folly::coro::Task { + co_await folly::coro::collectAll( + waitForEvent( + endpoint.get(), 2000ms), // NOLINT + folly::coro::co_invoke( + [this, + endpoint, + targetDescriptor, + disconnectMode]() -> folly::coro::Task { + co_return co_await this + ->doGracefullyDisconnectAsync( + targetDescriptor, disconnectMode); + })); + })); } catch (const std::exception& err) { SLogger::error( "Caught exception in gracefullyDisconnect " + diff --git a/packages/streamr-dht/modules/dht/discovery/PeerDiscovery.cppm b/packages/streamr-dht/modules/dht/discovery/PeerDiscovery.cppm index e10d51e6..99cf1a79 100644 --- a/packages/streamr-dht/modules/dht/discovery/PeerDiscovery.cppm +++ b/packages/streamr-dht/modules/dht/discovery/PeerDiscovery.cppm @@ -301,19 +301,17 @@ private: std::vector> fetches; fetches.reserve(nodes.size() + randomNodes.size()); for (const auto& node : nodes) { - fetches.push_back( - folly::coro::co_invoke( - [self, node, localNodeId]() -> folly::coro::Task { - co_await self->fetchAndAddContacts(node, localNodeId); - })); + fetches.push_back(folly::coro::co_invoke( + [self, node, localNodeId]() -> folly::coro::Task { + co_await self->fetchAndAddContacts(node, localNodeId); + })); } for (const auto& node : randomNodes) { - fetches.push_back( - folly::coro::co_invoke( - [self, node]() -> folly::coro::Task { - co_await self->fetchAndAddContacts( - node, Identifiers::createRandomDhtAddress()); - })); + fetches.push_back(folly::coro::co_invoke( + [self, node]() -> folly::coro::Task { + co_await self->fetchAndAddContacts( + node, Identifiers::createRandomDhtAddress()); + })); } // collectAllRange, but each fetch swallows its own failure — the TS // uses Promise.allSettled, so one failed remote does not abort the @@ -364,19 +362,15 @@ public: std::vector> joins; joins.reserve(entryPoints.size()); for (const auto& entryPoint : entryPoints) { - joins.push_back( - folly::coro::co_invoke( - [self, - entryPoint, - &contactedPeers, - &distantJoinOptions, - retry]() -> folly::coro::Task { - co_await self->joinThroughEntryPoint( - entryPoint, - contactedPeers, - distantJoinOptions, - retry); - })); + joins.push_back(folly::coro::co_invoke( + [self, + entryPoint, + &contactedPeers, + &distantJoinOptions, + retry]() -> folly::coro::Task { + co_await self->joinThroughEntryPoint( + entryPoint, contactedPeers, distantJoinOptions, retry); + })); } co_await folly::coro::collectAllRange(std::move(joins)); } @@ -401,7 +395,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); @@ -417,7 +411,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()); } } diff --git a/packages/streamr-dht/test/integration/ConnectionLockingTest.cpp b/packages/streamr-dht/test/integration/ConnectionLockingTest.cpp index 4251d4b3..108c25fa 100644 --- a/packages/streamr-dht/test/integration/ConnectionLockingTest.cpp +++ b/packages/streamr-dht/test/integration/ConnectionLockingTest.cpp @@ -117,8 +117,8 @@ class ConnectionLockingTest : public ::testing::Test { TEST_F(ConnectionLockingTest, CanLockConnections) { rtc::InitLogger(rtc::LogLevel::Verbose); SLogger::trace("In the beginning"); - auto connectionManager1 = createConnectionManager( - DefaultConnectorFacadeOptions{ + auto connectionManager1 = + createConnectionManager(DefaultConnectorFacadeOptions{ .transport = *mockConnectorTransport1, .websocketHost = "127.0.0.1", .websocketPortRange = @@ -129,8 +129,8 @@ TEST_F(ConnectionLockingTest, CanLockConnections) { SLogger::info("Starting connection manager 1"); connectionManager1->start(); - auto connectionManager2 = createConnectionManager( - DefaultConnectorFacadeOptions{ + auto connectionManager2 = + createConnectionManager(DefaultConnectorFacadeOptions{ .transport = *mockConnectorTransport2, .websocketHost = "127.0.0.1", .websocketPortRange = @@ -144,7 +144,8 @@ TEST_F(ConnectionLockingTest, CanLockConnections) { auto tmpMockPeerDescriptor2 = mockPeerDescriptor2; auto tmpMockPeerDescriptor1 = mockPeerDescriptor1; SLogger::trace("before lockConnection() start"); - connectionManager1->lockConnection(mockPeerDescriptor2, LockID("testLock")); + streamr::utils::blockingWait(connectionManager1->lockConnection( + mockPeerDescriptor2, LockID("testLock"))); SLogger::trace("lockConnection done"); std::function condition = [&connectionManager2, @@ -165,8 +166,8 @@ TEST_F(ConnectionLockingTest, CanLockConnections) { TEST_F(ConnectionLockingTest, LockingBothWays) { rtc::InitLogger(rtc::LogLevel::Verbose); SLogger::trace("In the beginning"); - auto connectionManager3 = createConnectionManager( - DefaultConnectorFacadeOptions{ + auto connectionManager3 = + createConnectionManager(DefaultConnectorFacadeOptions{ .transport = *mockConnectorTransport3, .websocketHost = "127.0.0.1", .websocketPortRange = @@ -179,8 +180,8 @@ TEST_F(ConnectionLockingTest, LockingBothWays) { SLogger::info("Starting connection manager 3"); connectionManager3->start(); - auto connectionManager4 = createConnectionManager( - DefaultConnectorFacadeOptions{ + auto connectionManager4 = + createConnectionManager(DefaultConnectorFacadeOptions{ .transport = *mockConnectorTransport4, .websocketHost = "127.0.0.1", .websocketPortRange = @@ -204,8 +205,8 @@ TEST_F(ConnectionLockingTest, LockingBothWays) { auto task = collect( [&]() { - connectionManager3->lockConnection( - mockPeerDescriptor4, LockID("testLock1")); + streamr::utils::blockingWait(connectionManager3->lockConnection( + mockPeerDescriptor4, LockID("testLock1"))); }, waitForCondition(condition) // NOLINT ); @@ -214,8 +215,8 @@ TEST_F(ConnectionLockingTest, LockingBothWays) { auto task2 = collect( [&]() { - connectionManager4->lockConnection( - mockPeerDescriptor3, LockID("testLock2")); + streamr::utils::blockingWait(connectionManager4->lockConnection( + mockPeerDescriptor3, LockID("testLock2"))); }, waitForCondition(condition) // NOLINT ); diff --git a/packages/streamr-trackerless-network/modules/logic/proxy/ProxyClient.cppm b/packages/streamr-trackerless-network/modules/logic/proxy/ProxyClient.cppm index 59b39755..2bd9c329 100644 --- a/packages/streamr-trackerless-network/modules/logic/proxy/ProxyClient.cppm +++ b/packages/streamr-trackerless-network/modules/logic/proxy/ProxyClient.cppm @@ -171,57 +171,50 @@ public: Identifiers::getNodeIdFromPeerDescriptor( options.localPeerDescriptor), 1000), // NOLINT - contentDeliveryRpcLocal( - ContentDeliveryRpcLocalOptions{ - .localPeerDescriptor = options.localPeerDescriptor, - .streamPartId = options.streamPartId, - .markAndCheckDuplicate = - [this]( - const MessageID& msg, - const std::optional& prev) { - std::scoped_lock lock(this->mutex); - return Utils::markAndCheckDuplicate( - this->duplicateDetectors, msg, prev); - }, - .broadcast = - [this]( - const StreamMessage& message, - const DhtAddress& previousNode) { - this->broadcast(message, previousNode); - }, - .onLeaveNotice = - [this]( - const DhtAddress& remoteNodeId, bool /*isLeaving*/) { - const auto contact = - this->neighbors.get(remoteNodeId); - if (contact.has_value()) { - this->onNodeDisconnected( - contact.value()->getPeerDescriptor()); - } - }, - - .markForInspection = [](const DhtAddress& nodeId, - const MessageID& message) {}, - .rpcCommunicator = this->rpcCommunicator}), - propagation( - PropagationOptions{ - .sendToNeighbor = - [this]( - const DhtAddress& neighborId, - const StreamMessage& msg) -> folly::coro::Task { - const auto remote = this->neighbors.get(neighborId); - if (remote.has_value()) { - co_await remote.value()->sendStreamMessage(msg); - } else { - throw std::runtime_error( - "Propagation target not found"); + contentDeliveryRpcLocal(ContentDeliveryRpcLocalOptions{ + .localPeerDescriptor = options.localPeerDescriptor, + .streamPartId = options.streamPartId, + .markAndCheckDuplicate = + [this]( + const MessageID& msg, + const std::optional& prev) { + std::scoped_lock lock(this->mutex); + return Utils::markAndCheckDuplicate( + this->duplicateDetectors, msg, prev); + }, + .broadcast = + [this]( + const StreamMessage& message, + const DhtAddress& previousNode) { + this->broadcast(message, previousNode); + }, + .onLeaveNotice = + [this](const DhtAddress& remoteNodeId, bool /*isLeaving*/) { + const auto contact = this->neighbors.get(remoteNodeId); + if (contact.has_value()) { + this->onNodeDisconnected( + contact.value()->getPeerDescriptor()); } }, - .minPropagationTargets = - options.minPropagationTargets.has_value() - ? options.minPropagationTargets.value() - : 2, - }), + + .markForInspection = [](const DhtAddress& nodeId, + const MessageID& message) {}, + .rpcCommunicator = this->rpcCommunicator}), + propagation(PropagationOptions{ + .sendToNeighbor = + [this](const DhtAddress& neighborId, const StreamMessage& msg) + -> folly::coro::Task { + const auto remote = this->neighbors.get(neighborId); + if (remote.has_value()) { + co_await remote.value()->sendStreamMessage(msg); + } else { + throw std::runtime_error("Propagation target not found"); + } + }, + .minPropagationTargets = options.minPropagationTargets.has_value() + ? options.minPropagationTargets.value() + : 2, + }), options(std::move(options)) {} private: @@ -393,14 +386,20 @@ public: std::current_exception(), peerDescriptor); } if (accepted) { - this->options.connectionLocker.lockConnection( - peerDescriptor, LockID{SERVICE_ID}); + // ProxyClient's connection routine is synchronous today (it + // blockingWaits its RPCs above); keep that behavior for the + // now-asynchronous locker API. Async-ifying this whole path + // is a separate task. + streamr::utils::blockingWait( + this->options.connectionLocker.lockConnection( + peerDescriptor, LockID{SERVICE_ID})); { std::scoped_lock lock(this->mutex); if (this->stopped) { - this->options.connectionLocker.unlockConnection( - peerDescriptor, LockID{SERVICE_ID}); + streamr::utils::blockingWait( + this->options.connectionLocker.unlockConnection( + peerDescriptor, LockID{SERVICE_ID})); return; } this->connections.emplace( @@ -464,8 +463,9 @@ public: streamr::utils::blockingWait(server.value()->leaveStreamPartNotice( this->options.streamPartId, false)); } - this->options.connectionLocker.unlockConnection( - peerDescriptor.value(), LockID{SERVICE_ID}); + streamr::utils::blockingWait( + this->options.connectionLocker.unlockConnection( + peerDescriptor.value(), LockID{SERVICE_ID})); this->neighbors.remove(nodeId); } @@ -552,8 +552,9 @@ public: return; } } - this->options.connectionLocker.unlockConnection( - peerDescriptor, LockID{SERVICE_ID}); + streamr::utils::blockingWait( + this->options.connectionLocker.unlockConnection( + peerDescriptor, LockID{SERVICE_ID})); this->neighbors.remove(nodeId); // Deviation from TS: no automatic reconnection attempt here. The // TS single retry always fails in practice (the proxy that @@ -590,8 +591,9 @@ public: // event thread parked on it may be the thread these sends need. auto allNeighbors = this->neighbors.getAll(); for (const auto& remote : allNeighbors) { - this->options.connectionLocker.unlockConnection( - remote->getPeerDescriptor(), LockID{SERVICE_ID}); + streamr::utils::blockingWait( + this->options.connectionLocker.unlockConnection( + remote->getPeerDescriptor(), LockID{SERVICE_ID})); streamr::utils::blockingWait(remote->leaveStreamPartNotice( this->options.streamPartId, false)); } diff --git a/packages/streamr-trackerless-network/test/utils/TestUtils.cppm b/packages/streamr-trackerless-network/test/utils/TestUtils.cppm index 052f81e6..cf939635 100644 --- a/packages/streamr-trackerless-network/test/utils/TestUtils.cppm +++ b/packages/streamr-trackerless-network/test/utils/TestUtils.cppm @@ -83,9 +83,8 @@ inline StreamMessage createStreamMessage( // identity (mirrors the streamr-dht test util of the same name). inline PeerDescriptor createMockPeerDescriptor() { PeerDescriptor descriptor; - descriptor.set_nodeid( - streamr::dht::Identifiers::getRawFromDhtAddress( - streamr::dht::Identifiers::createRandomDhtAddress())); + descriptor.set_nodeid(streamr::dht::Identifiers::getRawFromDhtAddress( + streamr::dht::Identifiers::createRandomDhtAddress())); descriptor.set_type(::dht::NodeType::NODEJS); return descriptor; } @@ -150,10 +149,14 @@ class MockConnectionLocker : public ConnectionLocker { public: ~MockConnectionLocker() override = default; - void lockConnection( - PeerDescriptor /*targetDescriptor*/, LockID /*lockId*/) override {} - void unlockConnection( - PeerDescriptor /*targetDescriptor*/, LockID /*lockId*/) override {} + folly::coro::Task lockConnection( + PeerDescriptor /*targetDescriptor*/, LockID /*lockId*/) override { + co_return; + } + folly::coro::Task unlockConnection( + PeerDescriptor /*targetDescriptor*/, LockID /*lockId*/) override { + co_return; + } void weakLockConnection( const DhtAddress& /*targetDescriptor*/, const LockID& /*lockId*/) override {} From 7a3a72abbd7058fb5051b3014c3aab0318f66974 Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Fri, 17 Jul 2026 09:46:19 +0300 Subject: [PATCH 5/8] DhtNode: drain the communicator's detached RPC coroutines in stop() MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The server-dispatch scope only drained in ~RoutingRpcCommunicator during member destruction — after later-declared members were already gone — so a straggler getClosestPeers dispatch resumed into freed memory (crash reports on the CI-sized pool; the teardown-drain-ordering rule). stop() now removes the transport listeners with offAndWait() (a Message handler mid-invocation can otherwise schedule one more dispatch past the drain) and calls rpcCommunicator->drainAsyncTasks() as its last step, after the leave notices and final replications that still send through the communicator. The RPC client/server APIs gate their scope-add paths on the drained flag: folly forbids add-after-join, so a late request now fails cleanly (client) or is dropped like any request to a stopped node (server). Co-Authored-By: Claude Fable 5 --- packages/streamr-dht/modules/dht/DhtNode.cppm | 171 +++++++++--------- .../modules/RpcCommunicatorClientApi.cppm | 111 ++++++------ .../modules/RpcCommunicatorServerApi.cppm | 23 ++- 3 files changed, 163 insertions(+), 142 deletions(-) diff --git a/packages/streamr-dht/modules/dht/DhtNode.cppm b/packages/streamr-dht/modules/dht/DhtNode.cppm index 6d78b4cb..9bae51f7 100644 --- a/packages/streamr-dht/modules/dht/DhtNode.cppm +++ b/packages/streamr-dht/modules/dht/DhtNode.cppm @@ -303,24 +303,22 @@ private: } void initPeerManager() { - this->peerManager = PeerManager::newInstance( - PeerManagerOptions{ - .numberOfNodesPerKBucket = - this->options.numberOfNodesPerKBucket, - .maxContactCount = this->options.maxContactCount, - .localNodeId = this->getNodeId(), - .localPeerDescriptor = *this->localPeerDescriptor, - .connectionLocker = this->connectionLockerShared, - .neighborPingLimit = this->options.neighborPingLimit, - .lockId = LockID{this->options.serviceId}, - .createDhtNodeRpcRemote = - [this](const PeerDescriptor& peerDescriptor) { - return this->createDhtNodeRpcRemote(peerDescriptor); - }, - .hasConnection = - [this](const DhtAddress& nodeId) { - return this->connectionsView->hasConnection(nodeId); - }}); + this->peerManager = PeerManager::newInstance(PeerManagerOptions{ + .numberOfNodesPerKBucket = this->options.numberOfNodesPerKBucket, + .maxContactCount = this->options.maxContactCount, + .localNodeId = this->getNodeId(), + .localPeerDescriptor = *this->localPeerDescriptor, + .connectionLocker = this->connectionLockerShared, + .neighborPingLimit = this->options.neighborPingLimit, + .lockId = LockID{this->options.serviceId}, + .createDhtNodeRpcRemote = + [this](const PeerDescriptor& peerDescriptor) { + return this->createDhtNodeRpcRemote(peerDescriptor); + }, + .hasConnection = + [this](const DhtAddress& nodeId) { + return this->connectionsView->hasConnection(nodeId); + }}); // StoreManager tracks the neighbourhood via nearbyContactAdded (TS // re-emits it as a DhtNode event first; wired straight through here). this->peerManager->on( @@ -366,17 +364,15 @@ private: const auto token = this->abortController.getSignal().getCancellationToken(); for (const auto& entryPoint : this->options.entryPoints) { - this->recoveryScope.add( - streamr::utils::co_withExecutor( - &this->recoveryExecutor, - streamr::utils::co_withCancellation( - token, - folly::coro::co_invoke( - [discovery, - entryPoint]() -> folly::coro::Task { - co_await discovery->rejoinDht( - entryPoint); - })))); + this->recoveryScope.add(streamr::utils::co_withExecutor( + &this->recoveryExecutor, + streamr::utils::co_withCancellation( + token, + folly::coro::co_invoke( + [discovery, + entryPoint]() -> folly::coro::Task { + co_await discovery->rejoinDht(entryPoint); + })))); } } }); @@ -589,32 +585,28 @@ public: this->initPeerManager(); - this->peerDiscovery = PeerDiscovery::newInstance( - PeerDiscoveryOptions{ - .localPeerDescriptor = *this->localPeerDescriptor, - .joinNoProgressLimit = this->options.joinNoProgressLimit, - .serviceId = this->options.serviceId, - .parallelism = this->options.joinParallelism, - .joinTimeout = this->options.dhtJoinTimeout, - .connectionLocker = this->connectionLockerShared, - .peerManager = *this->peerManager, - .abortSignal = this->abortController.getSignal(), - .createDhtNodeRpcRemote = - [this](const PeerDescriptor& peerDescriptor) { - return this->createDhtNodeRpcRemote(peerDescriptor); - }}); - this->router = Router::newInstance( - RouterOptions{ - .rpcCommunicator = *this->rpcCommunicator, - .localPeerDescriptor = *this->localPeerDescriptor, - .handleMessage = - [this](const Message& message) { - this->handleMessageFromRouter(message); - }, - .getConnections = - [this]() { - return this->connectionsView->getConnections(); - }}); + this->peerDiscovery = PeerDiscovery::newInstance(PeerDiscoveryOptions{ + .localPeerDescriptor = *this->localPeerDescriptor, + .joinNoProgressLimit = this->options.joinNoProgressLimit, + .serviceId = this->options.serviceId, + .parallelism = this->options.joinParallelism, + .joinTimeout = this->options.dhtJoinTimeout, + .connectionLocker = this->connectionLockerShared, + .peerManager = *this->peerManager, + .abortSignal = this->abortController.getSignal(), + .createDhtNodeRpcRemote = + [this](const PeerDescriptor& peerDescriptor) { + return this->createDhtNodeRpcRemote(peerDescriptor); + }}); + this->router = Router::newInstance(RouterOptions{ + .rpcCommunicator = *this->rpcCommunicator, + .localPeerDescriptor = *this->localPeerDescriptor, + .handleMessage = + [this](const Message& message) { + this->handleMessageFromRouter(message); + }, + .getConnections = + [this]() { return this->connectionsView->getConnections(); }}); auto routerPtr = this->router; this->recursiveOperationManager = RecursiveOperationManager::newInstance( @@ -649,30 +641,27 @@ public: [routerPtr](const std::string& requestId) { routerPtr->addToDuplicateDetector(requestId); }}); - this->storeManager = StoreManager::newInstance( - StoreManagerOptions{ - .rpcCommunicator = *this->rpcCommunicator, - .localPeerDescriptor = *this->localPeerDescriptor, - .localDataStore = this->localDataStore, - .serviceId = this->options.serviceId, - .highestTtl = this->options.storeHighestTtl, - .redundancyFactor = this->options.storageRedundancyFactor, - .getNeighbors = - [this]() { return this->getNeighborDescriptors(); }, - .createRpcRemote = - [this](const PeerDescriptor& contact) { - return std::make_shared( - *this->localPeerDescriptor, - contact, - StoreRpcClient(*this->rpcCommunicator), - this->options.rpcRequestTimeout); - }, - .executeRecursiveOperation = - [this]( - const DhtAddress& key, RecursiveOperation operation) { - return this->recursiveOperationManager->execute( - key, operation); - }}); + this->storeManager = StoreManager::newInstance(StoreManagerOptions{ + .rpcCommunicator = *this->rpcCommunicator, + .localPeerDescriptor = *this->localPeerDescriptor, + .localDataStore = this->localDataStore, + .serviceId = this->options.serviceId, + .highestTtl = this->options.storeHighestTtl, + .redundancyFactor = this->options.storageRedundancyFactor, + .getNeighbors = [this]() { return this->getNeighborDescriptors(); }, + .createRpcRemote = + [this](const PeerDescriptor& contact) { + return std::make_shared( + *this->localPeerDescriptor, + contact, + StoreRpcClient(*this->rpcCommunicator), + this->options.rpcRequestTimeout); + }, + .executeRecursiveOperation = + [this](const DhtAddress& key, RecursiveOperation operation) { + return this->recursiveOperationManager->execute( + key, operation); + }}); this->bindRpcLocalMethods(); // Set only after everything above succeeded: if start() throws // (e.g. the websocket server port is taken), transportPtr is still @@ -730,9 +719,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(this->messageToken); - this->transportPtr->off(this->connectedToken); - this->transportPtr->off(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( + this->messageToken); + this->transportPtr->offAndWait(this->connectedToken); + this->transportPtr->offAndWait(this->disconnectedToken); streamr::utils::blockingWait(this->storeManager->destroy()); this->localDataStore.clear(); if (this->peerManager) { @@ -753,6 +748,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 ------------------------------------------------------ diff --git a/packages/streamr-proto-rpc/modules/RpcCommunicatorClientApi.cppm b/packages/streamr-proto-rpc/modules/RpcCommunicatorClientApi.cppm index 3dc2d142..6b15f6de 100644 --- a/packages/streamr-proto-rpc/modules/RpcCommunicatorClientApi.cppm +++ b/packages/streamr-proto-rpc/modules/RpcCommunicatorClientApi.cppm @@ -228,6 +228,13 @@ public: const CallContextType& callContext, std::optional timeout = std::nullopt) { SLogger::trace("request(): methodName:", methodName); + if (mDrained) { + // A drained scope must not be add()ed to (folly forbids + // add-after-join); a request after drainAsyncTasks() is a + // stop()-ordering bug in the caller — fail it cleanly. + throw RpcClientError( + "request() called on a drained RpcCommunicator: " + methodName); + } std::chrono::milliseconds timeoutValue = mRpcRequestTimeout; if (timeout.has_value()) { @@ -254,31 +261,27 @@ public: // scope task settles. auto&& contract = folly::coro::makePromiseContract(); - mScope.add( - streamr::utils::co_withExecutor( - &streamr::utils::SharedExecutors::worker(), - folly::coro::co_invoke( - [requestMessage, - callContext, - timeoutValue, - this, - promise = std::move(contract.first)]() mutable - -> folly::coro::Task { - try { - auto ongoingRequest = - this->makeRpcRequest( - requestMessage, callContext); - promise.setValue( - co_await folly::coro::timeout( - std::move( - ongoingRequest->getFuture()), - timeoutValue)); - } catch (...) { - promise.setException( - folly::exception_wrapper( - std::current_exception())); - } - }))); + mScope.add(streamr::utils::co_withExecutor( + &streamr::utils::SharedExecutors::worker(), + folly::coro::co_invoke( + [requestMessage, + callContext, + timeoutValue, + this, + promise = std::move(contract.first)]() mutable + -> folly::coro::Task { + try { + auto ongoingRequest = + this->makeRpcRequest( + requestMessage, callContext); + promise.setValue(co_await folly::coro::timeout( + std::move(ongoingRequest->getFuture()), + timeoutValue)); + } catch (...) { + promise.setException(folly::exception_wrapper( + std::current_exception())); + } + }))); try { co_return co_await folly::coro::detachOnCancel( @@ -310,6 +313,12 @@ public: const CallContextType& callContext, std::optional timeout = std::nullopt) { SLogger::trace("notify() notificationName:", notificationName); + if (mDrained) { + // See request(): no adds after the scope drain. + throw RpcClientError( + "notify() called on a drained RpcCommunicator: " + + std::string(notificationName)); + } std::chrono::milliseconds timeoutValue = mRpcRequestTimeout; if (timeout.has_value()) { @@ -328,33 +337,31 @@ public: // the communicator owner), so it runs as an mScope task — the // scope drain in the destructor replaces the former private // executor's join. - mScope.add( - streamr::utils::co_withExecutor( - &streamr::utils::SharedExecutors::worker(), - folly::coro::co_invoke( - [requestMessage, - callContext, - promise = std::move(promiseContract.first), - outgoingMessageCallback = - mOutgoingMessageCallback]() mutable - -> folly::coro::Task { - try { - outgoingMessageCallback( - requestMessage, - requestMessage.requestid(), - callContext); - promise.setValue(); - } catch ( - const std::exception& clientSideException) { - SLogger::debug( - "Error when calling outgoing message callback from client for sending notification", - clientSideException.what()); - promise.setException(RpcClientError( - "Error when calling outgoing message callback from client for sending notification", - clientSideException.what())); - } - co_return; - }))); + mScope.add(streamr::utils::co_withExecutor( + &streamr::utils::SharedExecutors::worker(), + folly::coro::co_invoke( + [requestMessage, + callContext, + promise = std::move(promiseContract.first), + outgoingMessageCallback = + mOutgoingMessageCallback]() mutable + -> folly::coro::Task { + try { + outgoingMessageCallback( + requestMessage, + requestMessage.requestid(), + callContext); + promise.setValue(); + } catch (const std::exception& clientSideException) { + SLogger::debug( + "Error when calling outgoing message callback from client for sending notification", + clientSideException.what()); + promise.setException(RpcClientError( + "Error when calling outgoing message callback from client for sending notification", + clientSideException.what())); + } + co_return; + }))); co_return co_await folly::coro::timeout( folly::coro::detachOnCancel(std::move(promiseContract.second)), timeoutValue); diff --git a/packages/streamr-proto-rpc/modules/RpcCommunicatorServerApi.cppm b/packages/streamr-proto-rpc/modules/RpcCommunicatorServerApi.cppm index 86edaeb5..8b62785f 100644 --- a/packages/streamr-proto-rpc/modules/RpcCommunicatorServerApi.cppm +++ b/packages/streamr-proto-rpc/modules/RpcCommunicatorServerApi.cppm @@ -188,15 +188,22 @@ private: // immediately, so the delivery thread is never blocked by handler work. void handleRequest( const RpcMessage& rpcMessage, const CallContextType& callContext) { + if (mDrained) { + // The owner has drained this scope during its stop(); a + // request arriving after that is dropped like any request to + // a stopped node (and folly forbids add-after-join). + SLogger::debug( + "handleRequest() after drainAsyncTasks(), dropping request"); + return; + } auto handler = mServerRegistry.getAsyncHandler(rpcMessage); - mScope.add( - streamr::utils::co_withExecutor( - &mSerialExecutor, - makeResponseTask( - std::move(handler), - mOutgoingMessageCallback, - rpcMessage, - callContext))); + mScope.add(streamr::utils::co_withExecutor( + &mSerialExecutor, + makeResponseTask( + std::move(handler), + mOutgoingMessageCallback, + rpcMessage, + callContext))); } void handleNotification( From 03a7f23093bd7143f5d93901e3367637ad4caff6 Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Fri, 17 Jul 2026 09:46:19 +0300 Subject: [PATCH 6/8] Websocket connectors: close pending connections outside the mutex MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both destroy() methods held the connector mutex across the pendingConnection->close(true) call-outs. The Disconnected fan-out re-enters connection and connector code, so this deadlocks ABBA against any thread that holds a connection's mutex and needs the connector's in its handler chain (sampled: main thread in WebsocketClientConnector::destroy -> PendingConnection::close -> Disconnected handler blocked on a connection mutex, an rtc close callback holding it and blocked on the connector mutex, a pool worker in connect() queued behind — the never-hold-locks-across-call-outs rule). Both now snapshot and clear their maps under the lock and make every call-out after releasing it. Co-Authored-By: Claude Fable 5 --- .../websocket/WebsocketClientConnector.cppm | 52 ++++---- .../websocket/WebsocketServerConnector.cppm | 117 +++++++++--------- 2 files changed, 90 insertions(+), 79 deletions(-) diff --git a/packages/streamr-dht/modules/connection/websocket/WebsocketClientConnector.cppm b/packages/streamr-dht/modules/connection/websocket/WebsocketClientConnector.cppm index 9776b916..ef29927f 100644 --- a/packages/streamr-dht/modules/connection/websocket/WebsocketClientConnector.cppm +++ b/packages/streamr-dht/modules/connection/websocket/WebsocketClientConnector.cppm @@ -67,24 +67,22 @@ public: explicit WebsocketClientConnector(WebsocketClientConnectorOptions&& options) : options(std::move(options)), - rpcLocal( - WebsocketClientConnectorRpcLocalOptions{ - .connect = [this](const PeerDescriptor& targetPeerDescriptor) - -> std::shared_ptr { - return this->connect(targetPeerDescriptor, std::nullopt); + rpcLocal(WebsocketClientConnectorRpcLocalOptions{ + .connect = [this](const PeerDescriptor& targetPeerDescriptor) + -> std::shared_ptr { + return this->connect(targetPeerDescriptor, std::nullopt); + }, + .hasConnection = [this](const DhtAddress& nodeId) -> bool { + std::scoped_lock lock(this->mutex); + return this->connectingHandshakers.contains(nodeId) || + this->options.hasConnection(nodeId); + }, + .onNewConnection = + [this]( + const std::shared_ptr& connection) { + return this->options.onNewConnection(connection); }, - .hasConnection = [this](const DhtAddress& nodeId) -> bool { - std::scoped_lock lock(this->mutex); - return this->connectingHandshakers.contains(nodeId) || - this->options.hasConnection(nodeId); - }, - .onNewConnection = - [this]( - const std::shared_ptr& - connection) { - return this->options.onNewConnection(connection); - }, - .abortSignal = this->abortController.getSignal()}) { + .abortSignal = this->abortController.getSignal()}) { this->options.rpcCommunicator .registerRpcNotification( "requestConnection", @@ -157,14 +155,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> 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(); } }; diff --git a/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm b/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm index 812dc1da..11baad7e 100644 --- a/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm +++ b/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm @@ -136,8 +136,8 @@ public: explicit WebsocketServerConnector(WebsocketServerConnectorOptions&& options) : options(std::move(options)), host(this->options.host) { if (this->options.portRange.has_value()) { - this->websocketServer = std::make_unique(std::move( - WebsocketServerConfig{ + this->websocketServer = std::make_unique( + std::move(WebsocketServerConfig{ .portRange = this->options.portRange.value(), .enableTls = this->options.serverEnableTls.value_or(false), .tlsCertificateFiles = this->options.tlsCertificateFiles, @@ -171,10 +171,9 @@ public: this->websocketServer) { this->websocketServer->on< websocketserverevents:: - Connected>([this]( - const std::shared_ptr< - WebsocketServerConnection>& - serverSocket) { + Connected>([this](const std::shared_ptr< + WebsocketServerConnection>& + serverSocket) { const auto resourceUrl = serverSocket->getResourceURL(); const auto action = getActionFromUrl(resourceUrl); SLogger::trace( @@ -352,34 +351,41 @@ public: void destroy() { SLogger::trace("WebsocketServerConnector::destroy() called"); + std::map> handshakers; + std::map> requests; + std::unique_ptr 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 @@ -416,35 +422,34 @@ private: this->ongoingConnectRequests.erase(nodeId); }); this->ongoingConnectRequests.emplace(nodeId, pendingConnection); - this->requestConnectionScope.add( - streamr::utils::co_withExecutor( - &this->requestConnectionExecutor, - folly::coro::co_invoke( - [this, localPeerDescriptor, targetPeerDescriptor]() - -> folly::coro::Task { - try { - WebsocketClientConnectorRpcClient client( - this->options.rpcCommunicator); - WebsocketClientConnectorRpcRemote remoteConnector( - PeerDescriptor(localPeerDescriptor), - PeerDescriptor(targetPeerDescriptor), - std::move(client)); - // Cancellable by destroy(): abort fires before the - // executor join drains this task. - co_await streamr::utils::co_withCancellation( - this->abortController.getSignal() - .getCancellationToken(), - remoteConnector.requestConnection()); - SLogger::trace( - "Sent WebsocketConnectionRequest notification" - " to peer"); - } catch (const std::exception& err) { - SLogger::debug( - "Failed to send WebsocketConnectionRequest" - " notification to peer " + - std::string(err.what())); - } - }))); + this->requestConnectionScope.add(streamr::utils::co_withExecutor( + &this->requestConnectionExecutor, + folly::coro::co_invoke( + [this, localPeerDescriptor, targetPeerDescriptor]() + -> folly::coro::Task { + try { + WebsocketClientConnectorRpcClient client( + this->options.rpcCommunicator); + WebsocketClientConnectorRpcRemote remoteConnector( + PeerDescriptor(localPeerDescriptor), + PeerDescriptor(targetPeerDescriptor), + std::move(client)); + // Cancellable by destroy(): abort fires before the + // executor join drains this task. + co_await streamr::utils::co_withCancellation( + this->abortController.getSignal() + .getCancellationToken(), + remoteConnector.requestConnection()); + SLogger::trace( + "Sent WebsocketConnectionRequest notification" + " to peer"); + } catch (const std::exception& err) { + SLogger::debug( + "Failed to send WebsocketConnectionRequest" + " notification to peer " + + std::string(err.what())); + } + }))); return pendingConnection; } From d79c601b99af7d6b66bec22ecd4680af9d4acdb8 Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Fri, 17 Jul 2026 10:09:24 +0300 Subject: [PATCH 7/8] clang-format: converge on the checker's clang-format (llvm 22) The previous commits were formatted with the system clang-format; this re-runs run-clang-format.py -i with the same binary lint.sh checks against, mostly reverting cosmetic reflow the wrong version introduced. Co-Authored-By: Claude Fable 5 --- .../modules/connection/ConnectionManager.cppm | 109 ++++++------- .../websocket/WebsocketClientConnector.cppm | 32 ++-- .../websocket/WebsocketServerConnector.cppm | 68 ++++---- packages/streamr-dht/modules/dht/DhtNode.cppm | 147 ++++++++++-------- .../modules/dht/discovery/PeerDiscovery.cppm | 42 ++--- .../modules/dht/store/StoreManager.cppm | 44 +++--- .../integration/ConnectionLockingTest.cpp | 16 +- .../modules/RpcCommunicatorClientApi.cppm | 98 ++++++------ .../modules/RpcCommunicatorServerApi.cppm | 15 +- .../modules/ContentDeliveryManager.cppm | 90 ++++++----- .../modules/logic/proxy/ProxyClient.cppm | 91 ++++++----- .../test/utils/TestUtils.cppm | 5 +- 12 files changed, 404 insertions(+), 353 deletions(-) diff --git a/packages/streamr-dht/modules/connection/ConnectionManager.cppm b/packages/streamr-dht/modules/connection/ConnectionManager.cppm index 33aae94b..e014933c 100644 --- a/packages/streamr-dht/modules/connection/ConnectionManager.cppm +++ b/packages/streamr-dht/modules/connection/ConnectionManager.cppm @@ -166,41 +166,43 @@ public: this->send(message, sendOptions); }, RpcCommunicatorOptions{.rpcRequestTimeout = 10s}), // NOLINT - connectionLockRpcLocal(ConnectionLockRpcLocalOptions{ - .addRemoteLocked = - [this](const DhtAddress& id, const LockID& lockId) { - this->locks.addRemoteLocked(id, lockId); - }, - .removeRemoteLocked = - [this](const DhtAddress& id, const LockID& lockId) { - this->locks.removeRemoteLocked(id, lockId); - }, - .closeConnection = - [this]( - const PeerDescriptor& peerDescriptor, - bool gracefulLeave, - const std::optional& reason) { - SLogger::debug("closeConnection() callback of RpcLocal"); - this->closeConnection( - peerDescriptor, gracefulLeave, reason); - }, - .getLocalPeerDescriptor = - [this]() { return this->getLocalPeerDescriptor(); }, - .setPrivate = - [this](const DhtAddress& id, bool isPrivate) { - if (!this->options.allowIncomingPrivateConnections) { + connectionLockRpcLocal( + ConnectionLockRpcLocalOptions{ + .addRemoteLocked = + [this](const DhtAddress& id, const LockID& lockId) { + this->locks.addRemoteLocked(id, lockId); + }, + .removeRemoteLocked = + [this](const DhtAddress& id, const LockID& lockId) { + this->locks.removeRemoteLocked(id, lockId); + }, + .closeConnection = + [this]( + const PeerDescriptor& peerDescriptor, + bool gracefulLeave, + const std::optional& reason) { SLogger::debug( - "node " + id + - " attempted to set a connection as private," - " but it is not allowed"); - return; - } - if (isPrivate) { - this->locks.addPrivate(id); - } else { - this->locks.removePrivate(id); - } - }}) { + "closeConnection() callback of RpcLocal"); + this->closeConnection( + peerDescriptor, gracefulLeave, reason); + }, + .getLocalPeerDescriptor = + [this]() { return this->getLocalPeerDescriptor(); }, + .setPrivate = + [this](const DhtAddress& id, bool isPrivate) { + if (!this->options.allowIncomingPrivateConnections) { + SLogger::debug( + "node " + id + + " attempted to set a connection as private," + " but it is not allowed"); + return; + } + if (isPrivate) { + this->locks.addPrivate(id); + } else { + this->locks.removePrivate(id); + } + }}) { SLogger::debug("ConnectionManager constructor start"); SLogger::info("ConnectionManager constructor"); this->connectorFacade = this->options.createConnectorFacade(); @@ -744,24 +746,27 @@ private: if (endpoint->isConnected()) { try { SLogger::debug("gracefullyDisconnect() calling blockingWait()"); - streamr::utils::blockingWait(folly::coro::co_invoke( - [this, - endpoint, - targetDescriptor = std::move(targetDescriptor), - disconnectMode]() -> folly::coro::Task { - co_await folly::coro::collectAll( - waitForEvent( - endpoint.get(), 2000ms), // NOLINT - folly::coro::co_invoke( - [this, - endpoint, - targetDescriptor, - disconnectMode]() -> folly::coro::Task { - co_return co_await this - ->doGracefullyDisconnectAsync( - targetDescriptor, disconnectMode); - })); - })); + streamr::utils::blockingWait( + folly::coro::co_invoke( + [this, + endpoint, + targetDescriptor = std::move(targetDescriptor), + disconnectMode]() -> folly::coro::Task { + co_await folly::coro::collectAll( + waitForEvent( + endpoint.get(), 2000ms), // NOLINT + folly::coro::co_invoke( + [this, + endpoint, + targetDescriptor, + disconnectMode]() + -> folly::coro::Task { + co_return co_await this + ->doGracefullyDisconnectAsync( + targetDescriptor, + disconnectMode); + })); + })); } catch (const std::exception& err) { SLogger::error( "Caught exception in gracefullyDisconnect " + diff --git a/packages/streamr-dht/modules/connection/websocket/WebsocketClientConnector.cppm b/packages/streamr-dht/modules/connection/websocket/WebsocketClientConnector.cppm index ef29927f..24108ad6 100644 --- a/packages/streamr-dht/modules/connection/websocket/WebsocketClientConnector.cppm +++ b/packages/streamr-dht/modules/connection/websocket/WebsocketClientConnector.cppm @@ -67,22 +67,24 @@ public: explicit WebsocketClientConnector(WebsocketClientConnectorOptions&& options) : options(std::move(options)), - rpcLocal(WebsocketClientConnectorRpcLocalOptions{ - .connect = [this](const PeerDescriptor& targetPeerDescriptor) - -> std::shared_ptr { - return this->connect(targetPeerDescriptor, std::nullopt); - }, - .hasConnection = [this](const DhtAddress& nodeId) -> bool { - std::scoped_lock lock(this->mutex); - return this->connectingHandshakers.contains(nodeId) || - this->options.hasConnection(nodeId); - }, - .onNewConnection = - [this]( - const std::shared_ptr& connection) { - return this->options.onNewConnection(connection); + rpcLocal( + WebsocketClientConnectorRpcLocalOptions{ + .connect = [this](const PeerDescriptor& targetPeerDescriptor) + -> std::shared_ptr { + return this->connect(targetPeerDescriptor, std::nullopt); }, - .abortSignal = this->abortController.getSignal()}) { + .hasConnection = [this](const DhtAddress& nodeId) -> bool { + std::scoped_lock lock(this->mutex); + return this->connectingHandshakers.contains(nodeId) || + this->options.hasConnection(nodeId); + }, + .onNewConnection = + [this]( + const std::shared_ptr& + connection) { + return this->options.onNewConnection(connection); + }, + .abortSignal = this->abortController.getSignal()}) { this->options.rpcCommunicator .registerRpcNotification( "requestConnection", diff --git a/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm b/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm index 11baad7e..2fc3acf1 100644 --- a/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm +++ b/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm @@ -136,8 +136,8 @@ public: explicit WebsocketServerConnector(WebsocketServerConnectorOptions&& options) : options(std::move(options)), host(this->options.host) { if (this->options.portRange.has_value()) { - this->websocketServer = std::make_unique( - std::move(WebsocketServerConfig{ + this->websocketServer = std::make_unique(std::move( + WebsocketServerConfig{ .portRange = this->options.portRange.value(), .enableTls = this->options.serverEnableTls.value_or(false), .tlsCertificateFiles = this->options.tlsCertificateFiles, @@ -171,9 +171,10 @@ public: this->websocketServer) { this->websocketServer->on< websocketserverevents:: - Connected>([this](const std::shared_ptr< - WebsocketServerConnection>& - serverSocket) { + Connected>([this]( + const std::shared_ptr< + WebsocketServerConnection>& + serverSocket) { const auto resourceUrl = serverSocket->getResourceURL(); const auto action = getActionFromUrl(resourceUrl); SLogger::trace( @@ -422,34 +423,35 @@ private: this->ongoingConnectRequests.erase(nodeId); }); this->ongoingConnectRequests.emplace(nodeId, pendingConnection); - this->requestConnectionScope.add(streamr::utils::co_withExecutor( - &this->requestConnectionExecutor, - folly::coro::co_invoke( - [this, localPeerDescriptor, targetPeerDescriptor]() - -> folly::coro::Task { - try { - WebsocketClientConnectorRpcClient client( - this->options.rpcCommunicator); - WebsocketClientConnectorRpcRemote remoteConnector( - PeerDescriptor(localPeerDescriptor), - PeerDescriptor(targetPeerDescriptor), - std::move(client)); - // Cancellable by destroy(): abort fires before the - // executor join drains this task. - co_await streamr::utils::co_withCancellation( - this->abortController.getSignal() - .getCancellationToken(), - remoteConnector.requestConnection()); - SLogger::trace( - "Sent WebsocketConnectionRequest notification" - " to peer"); - } catch (const std::exception& err) { - SLogger::debug( - "Failed to send WebsocketConnectionRequest" - " notification to peer " + - std::string(err.what())); - } - }))); + this->requestConnectionScope.add( + streamr::utils::co_withExecutor( + &this->requestConnectionExecutor, + folly::coro::co_invoke( + [this, localPeerDescriptor, targetPeerDescriptor]() + -> folly::coro::Task { + try { + WebsocketClientConnectorRpcClient client( + this->options.rpcCommunicator); + WebsocketClientConnectorRpcRemote remoteConnector( + PeerDescriptor(localPeerDescriptor), + PeerDescriptor(targetPeerDescriptor), + std::move(client)); + // Cancellable by destroy(): abort fires before the + // executor join drains this task. + co_await streamr::utils::co_withCancellation( + this->abortController.getSignal() + .getCancellationToken(), + remoteConnector.requestConnection()); + SLogger::trace( + "Sent WebsocketConnectionRequest notification" + " to peer"); + } catch (const std::exception& err) { + SLogger::debug( + "Failed to send WebsocketConnectionRequest" + " notification to peer " + + std::string(err.what())); + } + }))); return pendingConnection; } diff --git a/packages/streamr-dht/modules/dht/DhtNode.cppm b/packages/streamr-dht/modules/dht/DhtNode.cppm index 9bae51f7..8092aab4 100644 --- a/packages/streamr-dht/modules/dht/DhtNode.cppm +++ b/packages/streamr-dht/modules/dht/DhtNode.cppm @@ -303,22 +303,24 @@ private: } void initPeerManager() { - this->peerManager = PeerManager::newInstance(PeerManagerOptions{ - .numberOfNodesPerKBucket = this->options.numberOfNodesPerKBucket, - .maxContactCount = this->options.maxContactCount, - .localNodeId = this->getNodeId(), - .localPeerDescriptor = *this->localPeerDescriptor, - .connectionLocker = this->connectionLockerShared, - .neighborPingLimit = this->options.neighborPingLimit, - .lockId = LockID{this->options.serviceId}, - .createDhtNodeRpcRemote = - [this](const PeerDescriptor& peerDescriptor) { - return this->createDhtNodeRpcRemote(peerDescriptor); - }, - .hasConnection = - [this](const DhtAddress& nodeId) { - return this->connectionsView->hasConnection(nodeId); - }}); + this->peerManager = PeerManager::newInstance( + PeerManagerOptions{ + .numberOfNodesPerKBucket = + this->options.numberOfNodesPerKBucket, + .maxContactCount = this->options.maxContactCount, + .localNodeId = this->getNodeId(), + .localPeerDescriptor = *this->localPeerDescriptor, + .connectionLocker = this->connectionLockerShared, + .neighborPingLimit = this->options.neighborPingLimit, + .lockId = LockID{this->options.serviceId}, + .createDhtNodeRpcRemote = + [this](const PeerDescriptor& peerDescriptor) { + return this->createDhtNodeRpcRemote(peerDescriptor); + }, + .hasConnection = + [this](const DhtAddress& nodeId) { + return this->connectionsView->hasConnection(nodeId); + }}); // StoreManager tracks the neighbourhood via nearbyContactAdded (TS // re-emits it as a DhtNode event first; wired straight through here). this->peerManager->on( @@ -364,15 +366,17 @@ private: const auto token = this->abortController.getSignal().getCancellationToken(); for (const auto& entryPoint : this->options.entryPoints) { - this->recoveryScope.add(streamr::utils::co_withExecutor( - &this->recoveryExecutor, - streamr::utils::co_withCancellation( - token, - folly::coro::co_invoke( - [discovery, - entryPoint]() -> folly::coro::Task { - co_await discovery->rejoinDht(entryPoint); - })))); + this->recoveryScope.add( + streamr::utils::co_withExecutor( + &this->recoveryExecutor, + streamr::utils::co_withCancellation( + token, + folly::coro::co_invoke( + [discovery, + entryPoint]() -> folly::coro::Task { + co_await discovery->rejoinDht( + entryPoint); + })))); } } }); @@ -585,28 +589,32 @@ public: this->initPeerManager(); - this->peerDiscovery = PeerDiscovery::newInstance(PeerDiscoveryOptions{ - .localPeerDescriptor = *this->localPeerDescriptor, - .joinNoProgressLimit = this->options.joinNoProgressLimit, - .serviceId = this->options.serviceId, - .parallelism = this->options.joinParallelism, - .joinTimeout = this->options.dhtJoinTimeout, - .connectionLocker = this->connectionLockerShared, - .peerManager = *this->peerManager, - .abortSignal = this->abortController.getSignal(), - .createDhtNodeRpcRemote = - [this](const PeerDescriptor& peerDescriptor) { - return this->createDhtNodeRpcRemote(peerDescriptor); - }}); - this->router = Router::newInstance(RouterOptions{ - .rpcCommunicator = *this->rpcCommunicator, - .localPeerDescriptor = *this->localPeerDescriptor, - .handleMessage = - [this](const Message& message) { - this->handleMessageFromRouter(message); - }, - .getConnections = - [this]() { return this->connectionsView->getConnections(); }}); + this->peerDiscovery = PeerDiscovery::newInstance( + PeerDiscoveryOptions{ + .localPeerDescriptor = *this->localPeerDescriptor, + .joinNoProgressLimit = this->options.joinNoProgressLimit, + .serviceId = this->options.serviceId, + .parallelism = this->options.joinParallelism, + .joinTimeout = this->options.dhtJoinTimeout, + .connectionLocker = this->connectionLockerShared, + .peerManager = *this->peerManager, + .abortSignal = this->abortController.getSignal(), + .createDhtNodeRpcRemote = + [this](const PeerDescriptor& peerDescriptor) { + return this->createDhtNodeRpcRemote(peerDescriptor); + }}); + this->router = Router::newInstance( + RouterOptions{ + .rpcCommunicator = *this->rpcCommunicator, + .localPeerDescriptor = *this->localPeerDescriptor, + .handleMessage = + [this](const Message& message) { + this->handleMessageFromRouter(message); + }, + .getConnections = + [this]() { + return this->connectionsView->getConnections(); + }}); auto routerPtr = this->router; this->recursiveOperationManager = RecursiveOperationManager::newInstance( @@ -641,27 +649,30 @@ public: [routerPtr](const std::string& requestId) { routerPtr->addToDuplicateDetector(requestId); }}); - this->storeManager = StoreManager::newInstance(StoreManagerOptions{ - .rpcCommunicator = *this->rpcCommunicator, - .localPeerDescriptor = *this->localPeerDescriptor, - .localDataStore = this->localDataStore, - .serviceId = this->options.serviceId, - .highestTtl = this->options.storeHighestTtl, - .redundancyFactor = this->options.storageRedundancyFactor, - .getNeighbors = [this]() { return this->getNeighborDescriptors(); }, - .createRpcRemote = - [this](const PeerDescriptor& contact) { - return std::make_shared( - *this->localPeerDescriptor, - contact, - StoreRpcClient(*this->rpcCommunicator), - this->options.rpcRequestTimeout); - }, - .executeRecursiveOperation = - [this](const DhtAddress& key, RecursiveOperation operation) { - return this->recursiveOperationManager->execute( - key, operation); - }}); + this->storeManager = StoreManager::newInstance( + StoreManagerOptions{ + .rpcCommunicator = *this->rpcCommunicator, + .localPeerDescriptor = *this->localPeerDescriptor, + .localDataStore = this->localDataStore, + .serviceId = this->options.serviceId, + .highestTtl = this->options.storeHighestTtl, + .redundancyFactor = this->options.storageRedundancyFactor, + .getNeighbors = + [this]() { return this->getNeighborDescriptors(); }, + .createRpcRemote = + [this](const PeerDescriptor& contact) { + return std::make_shared( + *this->localPeerDescriptor, + contact, + StoreRpcClient(*this->rpcCommunicator), + this->options.rpcRequestTimeout); + }, + .executeRecursiveOperation = + [this]( + const DhtAddress& key, RecursiveOperation operation) { + return this->recursiveOperationManager->execute( + key, operation); + }}); this->bindRpcLocalMethods(); // Set only after everything above succeeded: if start() throws // (e.g. the websocket server port is taken), transportPtr is still diff --git a/packages/streamr-dht/modules/dht/discovery/PeerDiscovery.cppm b/packages/streamr-dht/modules/dht/discovery/PeerDiscovery.cppm index 99cf1a79..982e464d 100644 --- a/packages/streamr-dht/modules/dht/discovery/PeerDiscovery.cppm +++ b/packages/streamr-dht/modules/dht/discovery/PeerDiscovery.cppm @@ -301,17 +301,19 @@ private: std::vector> fetches; fetches.reserve(nodes.size() + randomNodes.size()); for (const auto& node : nodes) { - fetches.push_back(folly::coro::co_invoke( - [self, node, localNodeId]() -> folly::coro::Task { - co_await self->fetchAndAddContacts(node, localNodeId); - })); + fetches.push_back( + folly::coro::co_invoke( + [self, node, localNodeId]() -> folly::coro::Task { + co_await self->fetchAndAddContacts(node, localNodeId); + })); } for (const auto& node : randomNodes) { - fetches.push_back(folly::coro::co_invoke( - [self, node]() -> folly::coro::Task { - co_await self->fetchAndAddContacts( - node, Identifiers::createRandomDhtAddress()); - })); + fetches.push_back( + folly::coro::co_invoke( + [self, node]() -> folly::coro::Task { + co_await self->fetchAndAddContacts( + node, Identifiers::createRandomDhtAddress()); + })); } // collectAllRange, but each fetch swallows its own failure — the TS // uses Promise.allSettled, so one failed remote does not abort the @@ -362,15 +364,19 @@ public: std::vector> joins; joins.reserve(entryPoints.size()); for (const auto& entryPoint : entryPoints) { - joins.push_back(folly::coro::co_invoke( - [self, - entryPoint, - &contactedPeers, - &distantJoinOptions, - retry]() -> folly::coro::Task { - co_await self->joinThroughEntryPoint( - entryPoint, contactedPeers, distantJoinOptions, retry); - })); + joins.push_back( + folly::coro::co_invoke( + [self, + entryPoint, + &contactedPeers, + &distantJoinOptions, + retry]() -> folly::coro::Task { + co_await self->joinThroughEntryPoint( + entryPoint, + contactedPeers, + distantJoinOptions, + retry); + })); } co_await folly::coro::collectAllRange(std::move(joins)); } diff --git a/packages/streamr-dht/modules/dht/store/StoreManager.cppm b/packages/streamr-dht/modules/dht/store/StoreManager.cppm index a9a6803f..b6915a65 100644 --- a/packages/streamr-dht/modules/dht/store/StoreManager.cppm +++ b/packages/streamr-dht/modules/dht/store/StoreManager.cppm @@ -131,14 +131,15 @@ private: auto self = this->sharedFromThis(); DataEntry entry = dataEntry; PeerDescriptor target = contact; - 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 { - co_await self->doReplicate(entry, target); - })))); + 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 { + co_await self->doReplicate(entry, target); + })))); } void registerLocalRpcMethods() { @@ -201,18 +202,21 @@ private: auto self = this->sharedFromThis(); DhtAddress key = dataKey; PeerDescriptor node = newNode; - 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 { - const auto dataEntries = - self->options.localDataStore.values(key); - for (const auto& dataEntry : dataEntries) { - co_await self->doReplicate(dataEntry, node); - } - })))); + 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 { + 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) { diff --git a/packages/streamr-dht/test/integration/ConnectionLockingTest.cpp b/packages/streamr-dht/test/integration/ConnectionLockingTest.cpp index 108c25fa..bba2bc83 100644 --- a/packages/streamr-dht/test/integration/ConnectionLockingTest.cpp +++ b/packages/streamr-dht/test/integration/ConnectionLockingTest.cpp @@ -117,8 +117,8 @@ class ConnectionLockingTest : public ::testing::Test { TEST_F(ConnectionLockingTest, CanLockConnections) { rtc::InitLogger(rtc::LogLevel::Verbose); SLogger::trace("In the beginning"); - auto connectionManager1 = - createConnectionManager(DefaultConnectorFacadeOptions{ + auto connectionManager1 = createConnectionManager( + DefaultConnectorFacadeOptions{ .transport = *mockConnectorTransport1, .websocketHost = "127.0.0.1", .websocketPortRange = @@ -129,8 +129,8 @@ TEST_F(ConnectionLockingTest, CanLockConnections) { SLogger::info("Starting connection manager 1"); connectionManager1->start(); - auto connectionManager2 = - createConnectionManager(DefaultConnectorFacadeOptions{ + auto connectionManager2 = createConnectionManager( + DefaultConnectorFacadeOptions{ .transport = *mockConnectorTransport2, .websocketHost = "127.0.0.1", .websocketPortRange = @@ -166,8 +166,8 @@ TEST_F(ConnectionLockingTest, CanLockConnections) { TEST_F(ConnectionLockingTest, LockingBothWays) { rtc::InitLogger(rtc::LogLevel::Verbose); SLogger::trace("In the beginning"); - auto connectionManager3 = - createConnectionManager(DefaultConnectorFacadeOptions{ + auto connectionManager3 = createConnectionManager( + DefaultConnectorFacadeOptions{ .transport = *mockConnectorTransport3, .websocketHost = "127.0.0.1", .websocketPortRange = @@ -180,8 +180,8 @@ TEST_F(ConnectionLockingTest, LockingBothWays) { SLogger::info("Starting connection manager 3"); connectionManager3->start(); - auto connectionManager4 = - createConnectionManager(DefaultConnectorFacadeOptions{ + auto connectionManager4 = createConnectionManager( + DefaultConnectorFacadeOptions{ .transport = *mockConnectorTransport4, .websocketHost = "127.0.0.1", .websocketPortRange = diff --git a/packages/streamr-proto-rpc/modules/RpcCommunicatorClientApi.cppm b/packages/streamr-proto-rpc/modules/RpcCommunicatorClientApi.cppm index 6b15f6de..7d5002dd 100644 --- a/packages/streamr-proto-rpc/modules/RpcCommunicatorClientApi.cppm +++ b/packages/streamr-proto-rpc/modules/RpcCommunicatorClientApi.cppm @@ -261,27 +261,31 @@ public: // scope task settles. auto&& contract = folly::coro::makePromiseContract(); - mScope.add(streamr::utils::co_withExecutor( - &streamr::utils::SharedExecutors::worker(), - folly::coro::co_invoke( - [requestMessage, - callContext, - timeoutValue, - this, - promise = std::move(contract.first)]() mutable - -> folly::coro::Task { - try { - auto ongoingRequest = - this->makeRpcRequest( - requestMessage, callContext); - promise.setValue(co_await folly::coro::timeout( - std::move(ongoingRequest->getFuture()), - timeoutValue)); - } catch (...) { - promise.setException(folly::exception_wrapper( - std::current_exception())); - } - }))); + mScope.add( + streamr::utils::co_withExecutor( + &streamr::utils::SharedExecutors::worker(), + folly::coro::co_invoke( + [requestMessage, + callContext, + timeoutValue, + this, + promise = std::move(contract.first)]() mutable + -> folly::coro::Task { + try { + auto ongoingRequest = + this->makeRpcRequest( + requestMessage, callContext); + promise.setValue( + co_await folly::coro::timeout( + std::move( + ongoingRequest->getFuture()), + timeoutValue)); + } catch (...) { + promise.setException( + folly::exception_wrapper( + std::current_exception())); + } + }))); try { co_return co_await folly::coro::detachOnCancel( @@ -337,31 +341,33 @@ public: // the communicator owner), so it runs as an mScope task — the // scope drain in the destructor replaces the former private // executor's join. - mScope.add(streamr::utils::co_withExecutor( - &streamr::utils::SharedExecutors::worker(), - folly::coro::co_invoke( - [requestMessage, - callContext, - promise = std::move(promiseContract.first), - outgoingMessageCallback = - mOutgoingMessageCallback]() mutable - -> folly::coro::Task { - try { - outgoingMessageCallback( - requestMessage, - requestMessage.requestid(), - callContext); - promise.setValue(); - } catch (const std::exception& clientSideException) { - SLogger::debug( - "Error when calling outgoing message callback from client for sending notification", - clientSideException.what()); - promise.setException(RpcClientError( - "Error when calling outgoing message callback from client for sending notification", - clientSideException.what())); - } - co_return; - }))); + mScope.add( + streamr::utils::co_withExecutor( + &streamr::utils::SharedExecutors::worker(), + folly::coro::co_invoke( + [requestMessage, + callContext, + promise = std::move(promiseContract.first), + outgoingMessageCallback = + mOutgoingMessageCallback]() mutable + -> folly::coro::Task { + try { + outgoingMessageCallback( + requestMessage, + requestMessage.requestid(), + callContext); + promise.setValue(); + } catch ( + const std::exception& clientSideException) { + SLogger::debug( + "Error when calling outgoing message callback from client for sending notification", + clientSideException.what()); + promise.setException(RpcClientError( + "Error when calling outgoing message callback from client for sending notification", + clientSideException.what())); + } + co_return; + }))); co_return co_await folly::coro::timeout( folly::coro::detachOnCancel(std::move(promiseContract.second)), timeoutValue); diff --git a/packages/streamr-proto-rpc/modules/RpcCommunicatorServerApi.cppm b/packages/streamr-proto-rpc/modules/RpcCommunicatorServerApi.cppm index 8b62785f..9519beb3 100644 --- a/packages/streamr-proto-rpc/modules/RpcCommunicatorServerApi.cppm +++ b/packages/streamr-proto-rpc/modules/RpcCommunicatorServerApi.cppm @@ -197,13 +197,14 @@ private: return; } auto handler = mServerRegistry.getAsyncHandler(rpcMessage); - mScope.add(streamr::utils::co_withExecutor( - &mSerialExecutor, - makeResponseTask( - std::move(handler), - mOutgoingMessageCallback, - rpcMessage, - callContext))); + mScope.add( + streamr::utils::co_withExecutor( + &mSerialExecutor, + makeResponseTask( + std::move(handler), + mOutgoingMessageCallback, + rpcMessage, + callContext))); } void handleNotification( diff --git a/packages/streamr-trackerless-network/modules/ContentDeliveryManager.cppm b/packages/streamr-trackerless-network/modules/ContentDeliveryManager.cppm index 212fa7b9..b969ad9f 100644 --- a/packages/streamr-trackerless-network/modules/ContentDeliveryManager.cppm +++ b/packages/streamr-trackerless-network/modules/ContentDeliveryManager.cppm @@ -357,39 +357,43 @@ public: SLogger::debug( "Manual rejoin required for stream part " + streamPartId); - this->joinScope.add(streamr::utils::co_withExecutor( - &this->joinExecutor, - streamr::utils::co_withCancellation( - this->joinCancellation.getToken(), - streamPartReconnect->reconnect()))); + this->joinScope.add( + streamr::utils::co_withExecutor( + &this->joinExecutor, + streamr::utils::co_withCancellation( + this->joinCancellation.getToken(), + streamPartReconnect->reconnect()))); } }); node->on( [this, streamPartId, peerDescriptorStoreManager]() { - this->joinScope.add(streamr::utils::co_withExecutor( - &this->joinExecutor, - streamr::utils::co_withCancellation( - this->joinCancellation.getToken(), - this->handleEntryPointLeave( - streamPartId, peerDescriptorStoreManager)))); + this->joinScope.add( + streamr::utils::co_withExecutor( + &this->joinExecutor, + streamr::utils::co_withCancellation( + this->joinCancellation.getToken(), + this->handleEntryPointLeave( + streamPartId, peerDescriptorStoreManager)))); }); // TS setImmediate(): detached bounded join. - this->joinScope.add(streamr::utils::co_withExecutor( - &this->joinExecutor, - streamr::utils::co_withCancellation( - this->joinCancellation.getToken(), - folly::coro::co_invoke( - [this, streamPartId, peerDescriptorStoreManager]() - -> folly::coro::Task { - try { - co_await this->startLayersAndJoinDht( - streamPartId, peerDescriptorStoreManager); - } catch (const std::exception& err) { - SLogger::warn( - "Failed to join to stream part " + - streamPartId + ": " + std::string(err.what())); - } - })))); + this->joinScope.add( + streamr::utils::co_withExecutor( + &this->joinExecutor, + streamr::utils::co_withCancellation( + this->joinCancellation.getToken(), + folly::coro::co_invoke( + [this, streamPartId, peerDescriptorStoreManager]() + -> folly::coro::Task { + try { + co_await this->startLayersAndJoinDht( + streamPartId, peerDescriptorStoreManager); + } catch (const std::exception& err) { + SLogger::warn( + "Failed to join to stream part " + + streamPartId + ": " + + std::string(err.what())); + } + })))); } folly::coro::Task setProxies( @@ -474,11 +478,12 @@ public: if (part->proxied) { continue; } - infos.push_back(StreamPartitionInfo{ - .id = streamPartId, - .controlLayerNeighbors = - part->discoveryLayerNode->getNeighbors(), - .contentDeliveryLayerNeighbors = part->node->getInfos()}); + infos.push_back( + StreamPartitionInfo{ + .id = streamPartId, + .controlLayerNeighbors = + part->discoveryLayerNode->getNeighbors(), + .contentDeliveryLayerNeighbors = part->node->getInfos()}); } return infos; } @@ -610,16 +615,17 @@ private: minNeighborCount) { auto networkSplitAvoidance = streamPart->networkSplitAvoidance; - this->joinScope.add(streamr::utils::co_withExecutor( - &this->joinExecutor, - streamr::utils::co_withCancellation( - this->joinCancellation.getToken(), - folly::coro::co_invoke( - [networkSplitAvoidance]() - -> folly::coro::Task { - co_await networkSplitAvoidance - ->avoidNetworkSplit(); - })))); + this->joinScope.add( + streamr::utils::co_withExecutor( + &this->joinExecutor, + streamr::utils::co_withCancellation( + this->joinCancellation.getToken(), + folly::coro::co_invoke( + [networkSplitAvoidance]() + -> folly::coro::Task { + co_await networkSplitAvoidance + ->avoidNetworkSplit(); + })))); } } } diff --git a/packages/streamr-trackerless-network/modules/logic/proxy/ProxyClient.cppm b/packages/streamr-trackerless-network/modules/logic/proxy/ProxyClient.cppm index 2bd9c329..9df25fca 100644 --- a/packages/streamr-trackerless-network/modules/logic/proxy/ProxyClient.cppm +++ b/packages/streamr-trackerless-network/modules/logic/proxy/ProxyClient.cppm @@ -171,50 +171,57 @@ public: Identifiers::getNodeIdFromPeerDescriptor( options.localPeerDescriptor), 1000), // NOLINT - contentDeliveryRpcLocal(ContentDeliveryRpcLocalOptions{ - .localPeerDescriptor = options.localPeerDescriptor, - .streamPartId = options.streamPartId, - .markAndCheckDuplicate = - [this]( - const MessageID& msg, - const std::optional& prev) { - std::scoped_lock lock(this->mutex); - return Utils::markAndCheckDuplicate( - this->duplicateDetectors, msg, prev); - }, - .broadcast = - [this]( - const StreamMessage& message, - const DhtAddress& previousNode) { - this->broadcast(message, previousNode); - }, - .onLeaveNotice = - [this](const DhtAddress& remoteNodeId, bool /*isLeaving*/) { - const auto contact = this->neighbors.get(remoteNodeId); - if (contact.has_value()) { - this->onNodeDisconnected( - contact.value()->getPeerDescriptor()); + contentDeliveryRpcLocal( + ContentDeliveryRpcLocalOptions{ + .localPeerDescriptor = options.localPeerDescriptor, + .streamPartId = options.streamPartId, + .markAndCheckDuplicate = + [this]( + const MessageID& msg, + const std::optional& prev) { + std::scoped_lock lock(this->mutex); + return Utils::markAndCheckDuplicate( + this->duplicateDetectors, msg, prev); + }, + .broadcast = + [this]( + const StreamMessage& message, + const DhtAddress& previousNode) { + this->broadcast(message, previousNode); + }, + .onLeaveNotice = + [this]( + const DhtAddress& remoteNodeId, bool /*isLeaving*/) { + const auto contact = + this->neighbors.get(remoteNodeId); + if (contact.has_value()) { + this->onNodeDisconnected( + contact.value()->getPeerDescriptor()); + } + }, + + .markForInspection = [](const DhtAddress& nodeId, + const MessageID& message) {}, + .rpcCommunicator = this->rpcCommunicator}), + propagation( + PropagationOptions{ + .sendToNeighbor = + [this]( + const DhtAddress& neighborId, + const StreamMessage& msg) -> folly::coro::Task { + const auto remote = this->neighbors.get(neighborId); + if (remote.has_value()) { + co_await remote.value()->sendStreamMessage(msg); + } else { + throw std::runtime_error( + "Propagation target not found"); } }, - - .markForInspection = [](const DhtAddress& nodeId, - const MessageID& message) {}, - .rpcCommunicator = this->rpcCommunicator}), - propagation(PropagationOptions{ - .sendToNeighbor = - [this](const DhtAddress& neighborId, const StreamMessage& msg) - -> folly::coro::Task { - const auto remote = this->neighbors.get(neighborId); - if (remote.has_value()) { - co_await remote.value()->sendStreamMessage(msg); - } else { - throw std::runtime_error("Propagation target not found"); - } - }, - .minPropagationTargets = options.minPropagationTargets.has_value() - ? options.minPropagationTargets.value() - : 2, - }), + .minPropagationTargets = + options.minPropagationTargets.has_value() + ? options.minPropagationTargets.value() + : 2, + }), options(std::move(options)) {} private: diff --git a/packages/streamr-trackerless-network/test/utils/TestUtils.cppm b/packages/streamr-trackerless-network/test/utils/TestUtils.cppm index cf939635..ce297fcc 100644 --- a/packages/streamr-trackerless-network/test/utils/TestUtils.cppm +++ b/packages/streamr-trackerless-network/test/utils/TestUtils.cppm @@ -83,8 +83,9 @@ inline StreamMessage createStreamMessage( // identity (mirrors the streamr-dht test util of the same name). inline PeerDescriptor createMockPeerDescriptor() { PeerDescriptor descriptor; - descriptor.set_nodeid(streamr::dht::Identifiers::getRawFromDhtAddress( - streamr::dht::Identifiers::createRandomDhtAddress())); + descriptor.set_nodeid( + streamr::dht::Identifiers::getRawFromDhtAddress( + streamr::dht::Identifiers::createRandomDhtAddress())); descriptor.set_type(::dht::NodeType::NODEJS); return descriptor; } From ff5f61744ca29766f04878ee7102fd2749e34fc6 Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Fri, 17 Jul 2026 12:28:35 +0300 Subject: [PATCH 8/8] CoroutineHelper: comment touch to invalidate the stale macOS CI cache MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The restored incremental-build cache on the macos-26 leg carried a pre-#92 BMI of this module and under-rebuilt: DiscoverySession and RingDiscoverySession failed to compile against the cancellation shims this branch added, while identical code built and passed on both Linux legs and locally. Touching the module interface forces the BMI (and its dependents) to rebuild — the documented remedy for this cache trap. Co-Authored-By: Claude Fable 5 --- packages/streamr-utils/modules/CoroutineHelper.cppm | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/packages/streamr-utils/modules/CoroutineHelper.cppm b/packages/streamr-utils/modules/CoroutineHelper.cppm index a9bb0f57..2f84aff2 100644 --- a/packages/streamr-utils/modules/CoroutineHelper.cppm +++ b/packages/streamr-utils/modules/CoroutineHelper.cppm @@ -73,6 +73,11 @@ using folly::Unit; export namespace streamr::utils { +// (Comment touch to invalidate stale CI build caches: the restored +// macOS cache carried a pre-cancellation-shim BMI of this module and +// ninja under-rebuilt, failing DiscoverySession/RingDiscoverySession +// compiles on symbols that exist — the documented incremental-cache +// trap.) // folly's blockingWait / co_withExecutor / co_withCancellation are // customization-point objects that cannot be re-exported by // using-declaration (internal linkage or conflicting redeclarations