From ba07aaa4c4e6a2c955ea8ae5cb6fa925918e70f4 Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Sun, 12 Jul 2026 09:40:14 +0300 Subject: [PATCH] Phase C2: neighbor discovery (Handshaker, NeighborFinder, NeighborUpdateManager) Ports the seven neighbor-discovery classes of the content-delivery layer from the TS pin, with all seven TS test files: - HandshakeRpcRemote/HandshakeRpcLocal/Handshaker: stream-part neighbor handshakes with parallel target selection from the four contact views and the interleaving protocol (a full node accepts the requester and hands one existing neighbor over). The TS detached interleave continuation becomes a time-bounded GuardedAsyncScope task. - NeighborFinder: repeated doFindNeighbors rounds (two concurrent chains, 250 ms apart) until minCount neighbors or contacts exhausted; INeighborFinder seam for the TS-style test doubles. - NeighborUpdateRpcRemote/RpcLocal/NeighborUpdateManager: periodic neighbor-list exchange with rtt recording (setRtt/getRtt added to ContentDeliveryRpcRemote) and removeMe pruning. - NodeList: public getLast(exclude) added (TS parity; only a static helper existed). - CoroutineHelper: export folly's collectAllTryRange (Promise.allSettled equivalent). - TestUtils: createMockContentDeliveryRpcRemote/createMockHandshakeRpcRemote factories over a transportless communicator whose sends fail fast. Local: trackerless-network unit 43/43 (21 new C2 tests). Co-Authored-By: Claude Fable 5 --- .../CMakeLists.txt | 7 + packages/streamr-trackerless-network/lint.sh | 5 +- .../logic/ContentDeliveryRpcRemote.cppm | 7 + .../modules/logic/NodeList.cppm | 6 + .../neighbor-discovery/HandshakeRpcLocal.cppm | 242 ++++++++++++ .../HandshakeRpcRemote.cppm | 134 +++++++ .../logic/neighbor-discovery/Handshaker.cppm | 347 ++++++++++++++++++ .../neighbor-discovery/NeighborFinder.cppm | 171 +++++++++ .../NeighborUpdateManager.cppm | 175 +++++++++ .../NeighborUpdateRpcLocal.cppm | 121 ++++++ .../NeighborUpdateRpcRemote.cppm | 90 +++++ .../test/unit/HandshakeRpcLocalTest.cpp | 178 +++++++++ .../test/unit/HandshakeRpcRemoteTest.cpp | 86 +++++ .../test/unit/HandshakerTest.cpp | 96 +++++ .../test/unit/HandshakesTest.cpp | 205 +++++++++++ .../test/unit/NeighborFinderTest.cpp | 93 +++++ .../test/unit/NeighborUpdateRpcLocalTest.cpp | 194 ++++++++++ .../test/unit/NeighborUpdateRpcRemoteTest.cpp | 88 +++++ .../test/utils/TestUtils.cppm | 61 +++ .../modules/CoroutineHelper.cppm | 1 + 20 files changed, 2306 insertions(+), 1 deletion(-) create mode 100644 packages/streamr-trackerless-network/modules/logic/neighbor-discovery/HandshakeRpcLocal.cppm create mode 100644 packages/streamr-trackerless-network/modules/logic/neighbor-discovery/HandshakeRpcRemote.cppm create mode 100644 packages/streamr-trackerless-network/modules/logic/neighbor-discovery/Handshaker.cppm create mode 100644 packages/streamr-trackerless-network/modules/logic/neighbor-discovery/NeighborFinder.cppm create mode 100644 packages/streamr-trackerless-network/modules/logic/neighbor-discovery/NeighborUpdateManager.cppm create mode 100644 packages/streamr-trackerless-network/modules/logic/neighbor-discovery/NeighborUpdateRpcLocal.cppm create mode 100644 packages/streamr-trackerless-network/modules/logic/neighbor-discovery/NeighborUpdateRpcRemote.cppm create mode 100644 packages/streamr-trackerless-network/test/unit/HandshakeRpcLocalTest.cpp create mode 100644 packages/streamr-trackerless-network/test/unit/HandshakeRpcRemoteTest.cpp create mode 100644 packages/streamr-trackerless-network/test/unit/HandshakerTest.cpp create mode 100644 packages/streamr-trackerless-network/test/unit/HandshakesTest.cpp create mode 100644 packages/streamr-trackerless-network/test/unit/NeighborFinderTest.cpp create mode 100644 packages/streamr-trackerless-network/test/unit/NeighborUpdateRpcLocalTest.cpp create mode 100644 packages/streamr-trackerless-network/test/unit/NeighborUpdateRpcRemoteTest.cpp diff --git a/packages/streamr-trackerless-network/CMakeLists.txt b/packages/streamr-trackerless-network/CMakeLists.txt index 8d6c1b0a..216bab89 100644 --- a/packages/streamr-trackerless-network/CMakeLists.txt +++ b/packages/streamr-trackerless-network/CMakeLists.txt @@ -175,6 +175,13 @@ if(NOT IOS AND STREAMR_MODULES_SUPPORTED) test/unit/NumberPairTest.cpp test/unit/StreamPartIdToDataKeyTest.cpp test/unit/TemporaryConnectionRpcLocalTest.cpp + test/unit/HandshakeRpcLocalTest.cpp + test/unit/HandshakeRpcRemoteTest.cpp + test/unit/HandshakerTest.cpp + test/unit/HandshakesTest.cpp + test/unit/NeighborFinderTest.cpp + test/unit/NeighborUpdateRpcLocalTest.cpp + test/unit/NeighborUpdateRpcRemoteTest.cpp ) target_include_directories(streamr-trackerless-network-test-unit diff --git a/packages/streamr-trackerless-network/lint.sh b/packages/streamr-trackerless-network/lint.sh index f5ab9cfc..7c7240ed 100755 --- a/packages/streamr-trackerless-network/lint.sh +++ b/packages/streamr-trackerless-network/lint.sh @@ -32,8 +32,11 @@ fi # false positive (std::string-vs-std::string mismatch on its own # locals). The compiler builds and runs both (unit suite green on every # platform); clang-format still checks them. +# HandshakerTest.cpp (phase C2) trips the same std-type unification +# false positive inside an included header ("no viable conversion from +# const std::string to __self_view"); the compiler builds and runs it. TESTFILES=$(find test -type f \( -name "*.hpp" -o -name "*.cpp" \) -not -path '*/ts-integration/*' | sort | uniq | tr '\n' ' ') -TIDY_TESTFILES=$(echo "$TESTFILES" | tr ' ' '\n' | grep -v 'test/unit/ContentDeliveryRpcRemoteTest.cpp' | grep -v 'test/unit/TemporaryConnectionRpcLocalTest.cpp' | tr '\n' ' ') +TIDY_TESTFILES=$(echo "$TESTFILES" | tr ' ' '\n' | grep -v 'test/unit/ContentDeliveryRpcRemoteTest.cpp' | grep -v 'test/unit/TemporaryConnectionRpcLocalTest.cpp' | grep -v 'test/unit/HandshakerTest.cpp' | tr '\n' ' ') echo "Running clangd-tidy on $TIDY_TESTFILES" clangd-tidy -p "$COMPILE_DB" $TIDY_TESTFILES < /dev/null diff --git a/packages/streamr-trackerless-network/modules/logic/ContentDeliveryRpcRemote.cppm b/packages/streamr-trackerless-network/modules/logic/ContentDeliveryRpcRemote.cppm index f47ac098..2b545248 100644 --- a/packages/streamr-trackerless-network/modules/logic/ContentDeliveryRpcRemote.cppm +++ b/packages/streamr-trackerless-network/modules/logic/ContentDeliveryRpcRemote.cppm @@ -34,6 +34,9 @@ using ::dht::PeerDescriptor; using ContentDeliveryRpcClient = streamr::protorpc::ContentDeliveryRpcClient; class ContentDeliveryRpcRemote : public RpcRemote { +private: + std::optional rtt; + public: ContentDeliveryRpcRemote( PeerDescriptor localPeerDescriptor, // NOLINT @@ -46,6 +49,10 @@ public: client, timeout) {} + void setRtt(int64_t rttMilliseconds) { this->rtt = rttMilliseconds; } + + [[nodiscard]] std::optional getRtt() const { return this->rtt; } + folly::coro::Task sendStreamMessage(StreamMessage msg) { auto options = this->formDhtRpcOptions({}); try { diff --git a/packages/streamr-trackerless-network/modules/logic/NodeList.cppm b/packages/streamr-trackerless-network/modules/logic/NodeList.cppm index 09db4717..34739651 100644 --- a/packages/streamr-trackerless-network/modules/logic/NodeList.cppm +++ b/packages/streamr-trackerless-network/modules/logic/NodeList.cppm @@ -175,6 +175,12 @@ public: : std::make_optional(included.back()); } + // TS getLast(exclude): the insertion-order last node not excluded. + [[nodiscard]] std::optional> + getLast(const std::vector& exclude) const { + return NodeList::getLast(this->nodes, exclude); + } + [[nodiscard]] std::vector> getAll() { return this->nodes | std::views::values | diff --git a/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/HandshakeRpcLocal.cppm b/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/HandshakeRpcLocal.cppm new file mode 100644 index 00000000..f937d119 --- /dev/null +++ b/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/HandshakeRpcLocal.cppm @@ -0,0 +1,242 @@ +// Module streamr.trackerlessnetwork.HandshakeRpcLocal +// Ported from packages/trackerless-network/src/content-delivery-layer/ +// neighbor-discovery/HandshakeRpcLocal.ts (v103.8.0-rc.3): the server +// side of stream-part neighbor handshakes, including the interleaving +// protocol (when full, accept the requester anyway and hand one existing +// neighbor over to it). +// +// Adaptation: TS fires the interleaveRequest continuation detached +// (".then() without await"); here it is a task on a GuardedAsyncScope +// drained on destruction. The task is time-bounded (the remote call +// carries interleaveRequestTimeout and swallows errors), satisfying the +// every-scope-task-must-be-bounded rule. +module; + +// Coroutine definitions need std::coroutine_traits declared in THIS +// translation unit; it cannot arrive through an imported BMI. +#include // IWYU pragma: keep + +#include +#include +#include +#include +#include +#include "packages/dht/protos/DhtRpc.pb.h" +#include "packages/network/protos/NetworkRpc.pb.h" + +export module streamr.trackerlessnetwork.HandshakeRpcLocal; + +import streamr.utils.CoroutineHelper; +import streamr.utils.GuardedAsyncScope; +import streamr.utils.SharedExecutors; +import streamr.trackerlessnetwork.ContentDeliveryRpcRemote; +import streamr.trackerlessnetwork.HandshakeRpcRemote; +import streamr.trackerlessnetwork.NodeList; +import streamr.dht.DhtCallContext; +import streamr.dht.Identifiers; +import streamr.dht.protos; +import streamr.logger.SLogger; +import streamr.utils.StreamPartID; + +// Hoisted (file scope, NOT exported); fully qualified because relative +// namespace names resolve differently at file scope than inside the +// package namespace. +using streamr::dht::DhtAddress; +using streamr::dht::DhtAddressRaw; +using streamr::dht::Identifiers; +using streamr::dht::rpcprotocol::DhtCallContext; +using streamr::logger::SLogger; +using streamr::trackerlessnetwork::ContentDeliveryRpcRemote; +using streamr::utils::GuardedAsyncScope; +using streamr::utils::StreamPartID; + +export namespace streamr::trackerlessnetwork::neighbordiscovery { + +using ::dht::PeerDescriptor; + +struct HandshakeRpcLocalOptions { + StreamPartID streamPartId; + NodeList& neighbors; + std::set& ongoingHandshakes; + std::set& ongoingInterleaves; + size_t maxNeighborCount; + std::function(const PeerDescriptor&)> + createRpcRemote; + std::function( + const PeerDescriptor&)> + createContentDeliveryRpcRemote; + std::function(PeerDescriptor, DhtAddress)> + handshakeWithInterleaving; +}; + +class HandshakeRpcLocal { +private: + HandshakeRpcLocalOptions options; + // Drains the detached interleave continuations (bounded by the + // remote call's timeout) before the members they mutate go away. + streamr::utils::SharedSerialExecutor interleaveExecutor{ + streamr::utils::SharedExecutors::worker()}; + GuardedAsyncScope interleaveScope; + +public: + explicit HandshakeRpcLocal(HandshakeRpcLocalOptions options) + : options(std::move(options)) {} + + ~HandshakeRpcLocal() { this->interleaveScope.close(); } + + StreamPartHandshakeResponse handshake( + const StreamPartHandshakeRequest& request, + const DhtCallContext& context) { + return this->handleRequest(request, context); + } + + folly::coro::Task interleaveRequest( + InterleaveRequest message, DhtCallContext context) { + const auto senderPeerDescriptor = + context.incomingSourceDescriptor.value(); + const auto remoteNodeId = + Identifiers::getNodeIdFromPeerDescriptor(senderPeerDescriptor); + InterleaveResponse response; + try { + co_await this->options.handshakeWithInterleaving( + message.interleavetargetdescriptor(), remoteNodeId); + this->options.neighbors.remove(remoteNodeId); + response.set_accepted(true); + } catch (const std::exception& err) { + SLogger::debug( + "interleaveRequest to " + + Identifiers::getNodeIdFromPeerDescriptor( + message.interleavetargetdescriptor()) + + " failed: " + std::string(err.what())); + response.set_accepted(false); + } + co_return response; + } + +private: + StreamPartHandshakeResponse handleRequest( + const StreamPartHandshakeRequest& request, + const DhtCallContext& context) { + const auto senderDescriptor = context.incomingSourceDescriptor.value(); + const auto senderNodeId = + Identifiers::getNodeIdFromPeerDescriptor(senderDescriptor); + std::vector interleaveNodeIds; + if (request.has_interleavenodeid()) { + interleaveNodeIds.push_back( + Identifiers::getDhtAddressFromRaw( + DhtAddressRaw{request.interleavenodeid()})); + } + if (this->options.ongoingInterleaves.contains(senderNodeId)) { + return rejectHandshake(request); + } + if (this->options.neighbors.has(senderNodeId) || + this->options.ongoingHandshakes.contains(senderNodeId)) { + return this->acceptHandshake(request, senderDescriptor); + } + if (this->options.neighbors.size() + + this->options.ongoingHandshakes.size() < + this->options.maxNeighborCount) { + return this->acceptHandshake(request, senderDescriptor); + } + if (this->options.neighbors.size(interleaveNodeIds) - + this->options.ongoingInterleaves.size() >= + 2 && + this->options.neighbors.size() <= this->options.maxNeighborCount) { + // Do not accept the handshake requests if the target neighbor + // count can potentially drop below 2 due to interleaving. + // This ensures that a stable number of connections is kept + // during high churn. + return this->acceptHandshakeWithInterleaving( + request, senderDescriptor); + } + return rejectHandshake(request); + } + + StreamPartHandshakeResponse acceptHandshake( + const StreamPartHandshakeRequest& request, + const PeerDescriptor& requester) { + StreamPartHandshakeResponse response; + response.set_requestid(request.requestid()); + response.set_accepted(true); + this->options.neighbors.add( + this->options.createContentDeliveryRpcRemote(requester)); + return response; + } + + static StreamPartHandshakeResponse rejectHandshake( + const StreamPartHandshakeRequest& request) { + StreamPartHandshakeResponse response; + response.set_requestid(request.requestid()); + response.set_accepted(false); + return response; + } + + StreamPartHandshakeResponse acceptHandshakeWithInterleaving( + const StreamPartHandshakeRequest& request, + const PeerDescriptor& requester) { + std::vector exclude; + for (const auto& id : request.neighbornodeids()) { + exclude.push_back( + Identifiers::getDhtAddressFromRaw(DhtAddressRaw{id})); + } + for (const auto& id : this->options.ongoingInterleaves) { + exclude.push_back(id); + } + exclude.push_back(Identifiers::getNodeIdFromPeerDescriptor(requester)); + if (request.has_interleavenodeid()) { + exclude.push_back( + Identifiers::getDhtAddressFromRaw( + DhtAddressRaw{request.interleavenodeid()})); + } + const auto last = this->options.neighbors.getLast(exclude); + std::optional lastPeerDescriptor; + if (last.has_value()) { + lastPeerDescriptor = last.value()->getPeerDescriptor(); + const auto nodeId = Identifiers::getNodeIdFromPeerDescriptor( + lastPeerDescriptor.value()); + const auto remote = + this->options.createRpcRemote(lastPeerDescriptor.value()); + this->options.ongoingInterleaves.insert(nodeId); + // TS runs this with then/catch instead of setImmediate to + // avoid changes in state; here it is a bounded scope task. + this->interleaveScope.add( + streamr::utils::co_withExecutor( + &this->interleaveExecutor, + folly::coro::co_invoke( + [this, + remote, + nodeId, + lastDescriptor = lastPeerDescriptor.value(), + requester]() -> folly::coro::Task { + try { + const auto response = + co_await remote->interleaveRequest( + requester); + // If accepted, remove the handed-over node from + // the target neighbors; otherwise keep it. + if (response.accepted()) { + this->options.neighbors.remove( + Identifiers:: + getNodeIdFromPeerDescriptor( + lastDescriptor)); + } + } catch (...) { + // no-op: interleaveRequest cannot reject + } + this->options.ongoingInterleaves.erase(nodeId); + }))); + } + this->options.neighbors.add( + this->options.createContentDeliveryRpcRemote(requester)); + StreamPartHandshakeResponse response; + response.set_requestid(request.requestid()); + response.set_accepted(true); + if (lastPeerDescriptor.has_value()) { + *response.mutable_interleavetargetdescriptor() = + lastPeerDescriptor.value(); + } + return response; + } +}; + +} // namespace streamr::trackerlessnetwork::neighbordiscovery diff --git a/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/HandshakeRpcRemote.cppm b/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/HandshakeRpcRemote.cppm new file mode 100644 index 00000000..2da5f2b1 --- /dev/null +++ b/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/HandshakeRpcRemote.cppm @@ -0,0 +1,134 @@ +// Module streamr.trackerlessnetwork.HandshakeRpcRemote +// Ported from packages/trackerless-network/src/content-delivery-layer/ +// neighbor-discovery/HandshakeRpcRemote.ts (v103.8.0-rc.3): the client +// side of stream-part neighbor handshakes. Both RPCs swallow errors and +// report them as not-accepted, matching the TS behavior. +module; + +// Coroutine definitions need std::coroutine_traits declared in THIS +// translation unit; it cannot arrive through an imported BMI. +#include // IWYU pragma: keep + +#include +#include +#include +#include +#include "packages/dht/protos/DhtRpc.pb.h" +#include "packages/network/protos/NetworkRpc.pb.h" + +export module streamr.trackerlessnetwork.HandshakeRpcRemote; + +import streamr.utils.CoroutineHelper; +import streamr.trackerlessnetwork.NetworkRpcClient; +import streamr.dht.DhtCallContext; +import streamr.dht.Identifiers; +import streamr.dht.RpcRemote; +import streamr.dht.protos; +import streamr.logger.SLogger; +import streamr.utils.StreamPartID; +import streamr.utils.Uuid; + +// Hoisted (file scope, NOT exported); fully qualified because relative +// namespace names resolve differently at file scope than inside the +// package namespace. +using streamr::dht::DhtAddress; +using streamr::dht::Identifiers; +using streamr::dht::contact::RpcRemote; +using streamr::dht::rpcprotocol::DhtCallContext; +using streamr::logger::SLogger; +using streamr::utils::StreamPartID; +using streamr::utils::Uuid; + +export namespace streamr::trackerlessnetwork::neighbordiscovery { + +using ::dht::PeerDescriptor; +using HandshakeRpcClient = + streamr::protorpc::HandshakeRpcClient; + +constexpr std::chrono::milliseconds interleaveRequestTimeout{10000}; + +struct HandshakeResponse { + bool accepted = false; + std::optional interleaveTargetDescriptor; +}; + +class HandshakeRpcRemote : public RpcRemote { +public: + HandshakeRpcRemote( + PeerDescriptor localPeerDescriptor, // NOLINT + PeerDescriptor remotePeerDescriptor, + HandshakeRpcClient client, + std::optional timeout = std::nullopt) + : RpcRemote( + std::move(localPeerDescriptor), + std::move(remotePeerDescriptor), + client, + timeout) {} + + folly::coro::Task handshake( + StreamPartID streamPartId, + std::vector neighborNodeIds, + std::optional concurrentHandshakeNodeId = std::nullopt, + std::optional interleaveNodeId = std::nullopt) { + StreamPartHandshakeRequest request; + request.set_streampartid(streamPartId); + request.set_requestid(Uuid::v4()); + for (const auto& id : neighborNodeIds) { + request.add_neighbornodeids(Identifiers::getRawFromDhtAddress(id)); + } + if (concurrentHandshakeNodeId.has_value()) { + request.set_concurrenthandshakenodeid( + Identifiers::getRawFromDhtAddress( + concurrentHandshakeNodeId.value())); + } + if (interleaveNodeId.has_value()) { + request.set_interleavenodeid( + Identifiers::getRawFromDhtAddress(interleaveNodeId.value())); + } + auto options = this->formDhtRpcOptions({}); + try { + const auto response = co_await this->getClient().handshake( + std::move(request), std::move(options)); + HandshakeResponse result{.accepted = response.accepted()}; + if (response.has_interleavetargetdescriptor()) { + result.interleaveTargetDescriptor = + response.interleavetargetdescriptor(); + } + co_return result; + } catch (const std::exception& err) { + SLogger::debug( + "handshake to " + + Identifiers::getNodeIdFromPeerDescriptor( + this->getPeerDescriptor()) + + " failed: " + std::string(err.what())); + co_return HandshakeResponse{.accepted = false}; + } + } + + folly::coro::Task interleaveRequest( + PeerDescriptor originatorDescriptor) { + InterleaveRequest request; + *request.mutable_interleavetargetdescriptor() = + std::move(originatorDescriptor); + DhtCallContext opts; + opts.connect = false; + auto options = this->formDhtRpcOptions(opts); + try { + co_return co_await this->getClient().interleaveRequest( + std::move(request), + std::move(options), + interleaveRequestTimeout); + } catch (const std::exception& err) { + SLogger::debug( + "interleaveRequest to " + + Identifiers::getNodeIdFromPeerDescriptor( + this->getPeerDescriptor()) + + " failed: " + std::string(err.what())); + InterleaveResponse response; + response.set_accepted(false); + co_return response; + } + } +}; + +} // namespace streamr::trackerlessnetwork::neighbordiscovery diff --git a/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/Handshaker.cppm b/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/Handshaker.cppm new file mode 100644 index 00000000..8c95f2dc --- /dev/null +++ b/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/Handshaker.cppm @@ -0,0 +1,347 @@ +// Module streamr.trackerlessnetwork.Handshaker +// Ported from packages/trackerless-network/src/content-delivery-layer/ +// neighbor-discovery/Handshaker.ts (v103.8.0-rc.3): selects handshake +// targets from the four contact views (ring left/right, Kademlia-nearby, +// random), runs up to two handshakes in parallel, and handles the +// interleaving protocol via HandshakeRpcLocal. +module; + +// Coroutine definitions need std::coroutine_traits declared in THIS +// translation unit; it cannot arrive through an imported BMI. +#include // IWYU pragma: keep + +#include +#include +#include +#include +#include +#include +#include +#include "packages/dht/protos/DhtRpc.pb.h" +#include "packages/network/protos/NetworkRpc.pb.h" + +export module streamr.trackerlessnetwork.Handshaker; + +import streamr.utils.CoroutineHelper; +import streamr.trackerlessnetwork.ContentDeliveryRpcRemote; +import streamr.trackerlessnetwork.HandshakeRpcLocal; +import streamr.trackerlessnetwork.HandshakeRpcRemote; +import streamr.trackerlessnetwork.NetworkRpcClient; +import streamr.trackerlessnetwork.NodeList; +import streamr.dht.DhtCallContext; +import streamr.dht.Identifiers; +import streamr.dht.ListeningRpcCommunicator; +import streamr.dht.protos; +import streamr.logger.SLogger; +import streamr.utils.StreamPartID; + +// Hoisted (file scope, NOT exported); fully qualified because relative +// namespace names resolve differently at file scope than inside the +// package namespace. +using streamr::dht::DhtAddress; +using streamr::dht::Identifiers; +using streamr::dht::rpcprotocol::DhtCallContext; +using streamr::dht::transport::ListeningRpcCommunicator; +using streamr::logger::SLogger; +using streamr::trackerlessnetwork::ContentDeliveryRpcRemote; +using streamr::utils::StreamPartID; + +export namespace streamr::trackerlessnetwork::neighbordiscovery { + +using ::dht::PeerDescriptor; +using streamr::trackerlessnetwork::ContentDeliveryRpcClient; + +constexpr size_t parallelHandshakeCount = 2; + +struct HandshakerOptions { + PeerDescriptor localPeerDescriptor; + StreamPartID streamPartId; + NodeList& neighbors; + NodeList& leftNodeView; + NodeList& rightNodeView; + NodeList& nearbyNodeView; + NodeList& randomNodeView; + ListeningRpcCommunicator& rpcCommunicator; + size_t maxNeighborCount; + std::set& ongoingHandshakes; + std::optional rpcRequestTimeout = std::nullopt; +}; + +class Handshaker { +private: + HandshakerOptions options; + std::set ongoingInterleaves; + HandshakeRpcLocal rpcLocal; + +public: + explicit Handshaker(HandshakerOptions options) + : options(std::move(options)), + rpcLocal( + HandshakeRpcLocalOptions{ + .streamPartId = this->options.streamPartId, + .neighbors = this->options.neighbors, + .ongoingHandshakes = this->options.ongoingHandshakes, + .ongoingInterleaves = this->ongoingInterleaves, + .maxNeighborCount = this->options.maxNeighborCount, + .createRpcRemote = + [this](const PeerDescriptor& target) { + return this->createRpcRemote(target); + }, + .createContentDeliveryRpcRemote = + [this](const PeerDescriptor& target) { + return this->createContentDeliveryRpcRemote(target); + }, + .handshakeWithInterleaving = + [this](PeerDescriptor target, DhtAddress remoteNodeId) + -> folly::coro::Task { + return this->handshakeWithInterleaving( + std::move(target), std::move(remoteNodeId)); + }}) { + this->options.rpcCommunicator + .registerRpcMethodAsync( + "interleaveRequest", + [this](InterleaveRequest request, DhtCallContext context) { + return this->rpcLocal.interleaveRequest( + std::move(request), std::move(context)); + }, + {.timeout = + static_cast(interleaveRequestTimeout.count())}); + this->options.rpcCommunicator.registerRpcMethod< + StreamPartHandshakeRequest, + StreamPartHandshakeResponse>( + "handshake", + [this]( + const StreamPartHandshakeRequest& request, + const DhtCallContext& context) { + return this->rpcLocal.handshake(request, context); + }); + } + + folly::coro::Task> attemptHandshakesOnContacts( + std::vector excludedIds) { + // TS TODO preserved: use an options option or named constant? + // or why the value 2? + if (this->options.neighbors.size() + + this->options.ongoingHandshakes.size() < + this->options.maxNeighborCount - 2) { + SLogger::trace("Attempting parallel handshakes with 2 targets"); + co_return co_await this->selectParallelTargetsAndHandshake( + std::move(excludedIds)); + } + if (this->options.neighbors.size() + + this->options.ongoingHandshakes.size() < + this->options.maxNeighborCount) { + SLogger::trace("Attempting handshake with new target"); + co_return co_await this->selectNewTargetAndHandshake( + std::move(excludedIds)); + } + co_return excludedIds; + } + + [[nodiscard]] const std::set& getOngoingHandshakes() const { + return this->options.ongoingHandshakes; + } + +private: + folly::coro::Task> + selectParallelTargetsAndHandshake(std::vector excludedIds) { + auto exclude = excludedIds; + for (const auto& id : this->options.neighbors.getIds()) { + exclude.push_back(id); + } + const auto targets = this->selectParallelTargets(exclude); + for (const auto& target : targets) { + this->options.ongoingHandshakes.insert( + Identifiers::getNodeIdFromPeerDescriptor( + target->getPeerDescriptor())); + } + co_return co_await this->doParallelHandshakes( + targets, std::move(exclude)); + } + + std::vector> selectParallelTargets( + const std::vector& excludedIds) { + // Insertion-ordered and deduplicated by node id (the TS version + // uses a Map for the same effect). + std::vector< + std::pair>> + targets; + const auto getExcludedIds = [&excludedIds, &targets]() { + auto ids = excludedIds; + for (const auto& [id, remote] : targets) { + ids.push_back(id); + } + return ids; + }; + const auto addTarget = + [&targets]( + const std::optional>& + candidate) { + if (!candidate.has_value()) { + return; + } + const auto nodeId = Identifiers::getNodeIdFromPeerDescriptor( + candidate.value()->getPeerDescriptor()); + const auto isKnown = + std::ranges::any_of(targets, [&nodeId](const auto& pair) { + return pair.first == nodeId; + }); + if (!isKnown) { + targets.emplace_back(nodeId, candidate.value()); + } + }; + + // Step 1: If no neighbors, try to find a WebSocket node first + if (this->options.neighbors.size() == 0) { + addTarget( + this->options.nearbyNodeView.getFirst(getExcludedIds(), true)); + } + // Step 2: Add left and right contacts from the ring + addTarget(this->options.leftNodeView.getFirst(getExcludedIds())); + addTarget(this->options.rightNodeView.getFirst(getExcludedIds())); + // Step 3: Add closest contact based on Kademlia metric if needed + if (targets.size() < parallelHandshakeCount) { + addTarget(this->options.nearbyNodeView.getFirst(getExcludedIds())); + } + // Step 4: Fill remaining slots with random contacts + while (targets.size() < parallelHandshakeCount) { + const auto random = + this->options.randomNodeView.getRandom(getExcludedIds()); + if (!random.has_value()) { + break; + } + addTarget(random); + } + + std::vector> remotes; + remotes.reserve(targets.size()); + for (const auto& [id, neighbor] : targets) { + remotes.push_back( + this->createRpcRemote(neighbor->getPeerDescriptor())); + } + return remotes; + } + + folly::coro::Task> doParallelHandshakes( + std::vector> targets, + std::vector excludedIds) { + std::vector> handshakes; + handshakes.reserve(targets.size()); + for (size_t i = 0; i < targets.size(); i++) { + const auto& otherNode = (i == 0) + ? ((targets.size() > 1) ? targets[1] : nullptr) + : targets[0]; + // TS TODO preserved: better check (currently this condition + // is always true) + std::optional otherNodeId = otherNode + ? std::optional( + Identifiers::getNodeIdFromPeerDescriptor( + otherNode->getPeerDescriptor())) + : std::nullopt; + handshakes.push_back( + this->handshakeWithTarget(targets[i], otherNodeId)); + } + const auto results = + co_await folly::coro::collectAllTryRange(std::move(handshakes)); + for (size_t i = 0; i < results.size(); i++) { + if (!results[i].hasValue() || !results[i].value()) { + excludedIds.push_back( + Identifiers::getNodeIdFromPeerDescriptor( + targets[i]->getPeerDescriptor())); + } + } + co_return excludedIds; + } + + folly::coro::Task> selectNewTargetAndHandshake( + std::vector excludedIds) { + auto exclude = excludedIds; + for (const auto& id : this->options.neighbors.getIds()) { + exclude.push_back(id); + } + auto target = this->options.leftNodeView.getFirst(exclude); + if (!target.has_value()) { + target = this->options.rightNodeView.getFirst(exclude); + } + if (!target.has_value()) { + target = this->options.nearbyNodeView.getFirst(exclude); + } + if (!target.has_value()) { + target = this->options.randomNodeView.getRandom(exclude); + } + if (target.has_value()) { + const auto accepted = co_await this->handshakeWithTarget( + this->createRpcRemote(target.value()->getPeerDescriptor()), + std::nullopt); + if (!accepted) { + excludedIds.push_back( + Identifiers::getNodeIdFromPeerDescriptor( + target.value()->getPeerDescriptor())); + } + } + co_return excludedIds; + } + + folly::coro::Task handshakeWithTarget( + std::shared_ptr target, + std::optional concurrentNodeId) { + const auto targetNodeId = Identifiers::getNodeIdFromPeerDescriptor( + target->getPeerDescriptor()); + this->options.ongoingHandshakes.insert(targetNodeId); + const auto result = co_await target->handshake( + this->options.streamPartId, + this->options.neighbors.getIds(), + concurrentNodeId); + if (result.accepted) { + this->options.neighbors.add(this->createContentDeliveryRpcRemote( + target->getPeerDescriptor())); + } + if (result.interleaveTargetDescriptor.has_value()) { + co_await this->handshakeWithInterleaving( + result.interleaveTargetDescriptor.value(), targetNodeId); + } + this->options.ongoingHandshakes.erase(targetNodeId); + co_return result.accepted; + } + + folly::coro::Task handshakeWithInterleaving( + PeerDescriptor target, DhtAddress remoteNodeId) { + const auto remote = this->createRpcRemote(target); + const auto targetNodeId = Identifiers::getNodeIdFromPeerDescriptor( + remote->getPeerDescriptor()); + this->options.ongoingHandshakes.insert(targetNodeId); + const auto result = co_await remote->handshake( + this->options.streamPartId, + this->options.neighbors.getIds(), + std::nullopt, + remoteNodeId); + if (result.accepted) { + this->options.neighbors.add(this->createContentDeliveryRpcRemote( + remote->getPeerDescriptor())); + } + this->options.ongoingHandshakes.erase(targetNodeId); + co_return result.accepted; + } + + std::shared_ptr createRpcRemote( + const PeerDescriptor& targetPeerDescriptor) { + HandshakeRpcClient client{this->options.rpcCommunicator}; + return std::make_shared( + this->options.localPeerDescriptor, + targetPeerDescriptor, + client, + this->options.rpcRequestTimeout); + } + + std::shared_ptr createContentDeliveryRpcRemote( + const PeerDescriptor& targetPeerDescriptor) { + ContentDeliveryRpcClient client{this->options.rpcCommunicator}; + return std::make_shared( + this->options.localPeerDescriptor, + targetPeerDescriptor, + client, + this->options.rpcRequestTimeout); + } +}; + +} // namespace streamr::trackerlessnetwork::neighbordiscovery diff --git a/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/NeighborFinder.cppm b/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/NeighborFinder.cppm new file mode 100644 index 00000000..df0824a8 --- /dev/null +++ b/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/NeighborFinder.cppm @@ -0,0 +1,171 @@ +// Module streamr.trackerlessnetwork.NeighborFinder +// Ported from packages/trackerless-network/src/content-delivery-layer/ +// neighbor-discovery/NeighborFinder.ts (v103.8.0-rc.3): drives repeated +// doFindNeighbors rounds (250 ms apart, two concurrent chains) until the +// neighbor list reaches minCount or the contact views are exhausted. +// +// Adaptation: the TS recursion via setAbortableTimeout becomes timer +// callbacks that enqueue the next round on a GuardedAsyncScope (executor- +// attached per the scope's contract); each round is bounded by +// doFindNeighbors itself, and stop() aborts the timers before draining +// the scope. +module; + +// Coroutine definitions need std::coroutine_traits declared in THIS +// translation unit; it cannot arrive through an imported BMI. +#include // IWYU pragma: keep + +#include +#include +#include +#include +#include + +export module streamr.trackerlessnetwork.NeighborFinder; + +import streamr.utils.CoroutineHelper; +import streamr.utils.AbortController; +import streamr.utils.AbortableTimers; +import streamr.utils.GuardedAsyncScope; +import streamr.utils.SharedExecutors; +import streamr.trackerlessnetwork.NodeList; +import streamr.dht.Identifiers; +import streamr.logger.SLogger; + +// Hoisted (file scope, NOT exported); fully qualified because relative +// namespace names resolve differently at file scope than inside the +// package namespace. +using streamr::dht::DhtAddress; +using streamr::logger::SLogger; +using streamr::utils::AbortableTimers; +using streamr::utils::AbortController; +using streamr::utils::GuardedAsyncScope; +using streamr::utils::SharedSerialExecutor; + +export namespace streamr::trackerlessnetwork::neighbordiscovery { + +constexpr std::chrono::milliseconds neighborFinderInitialWait{100}; +constexpr std::chrono::milliseconds neighborFinderInterval{250}; + +// Small seam so NeighborUpdateRpcLocal / NeighborUpdateManager can be +// unit-tested with a counting double (the TS tests inject +// `{ start: jest.fn() }`). +class INeighborFinder { +public: + virtual ~INeighborFinder() = default; + virtual void start(std::vector excluded) = 0; + void start() { this->start({}); } + virtual void stop() = 0; + [[nodiscard]] virtual bool isRunning() const = 0; +}; + +struct NeighborFinderOptions { + NodeList& neighbors; + NodeList& nearbyNodeView; + NodeList& leftNodeView; + NodeList& rightNodeView; + NodeList& randomNodeView; + std::function>( + std::vector)> + doFindNeighbors; + size_t minCount; +}; + +class NeighborFinder : public INeighborFinder { +private: + NeighborFinderOptions options; + AbortController abortController; + // Serial view so the rounds of the two chains never overlap; the + // scope drain in stop() is bounded because doFindNeighbors is (its + // RPCs carry timeouts). + SharedSerialExecutor executor{streamr::utils::SharedExecutors::worker()}; + GuardedAsyncScope scope; + std::atomic running = false; + +public: + explicit NeighborFinder(NeighborFinderOptions options) + : options(std::move(options)) {} + + ~NeighborFinder() override { this->stop(); } + + using INeighborFinder::start; + + void start(std::vector excluded) override { + if (this->running.exchange(true)) { + return; + } + AbortableTimers::setAbortableTimeout( + [this, excluded]() { + // TS runs two concurrent find chains. + this->scheduleFindNeighbors(excluded); + this->scheduleFindNeighbors(excluded); + }, + neighborFinderInitialWait, + this->abortController.getSignal()); + } + + void stop() override { + if (!this->running.exchange(false)) { + return; + } + this->abortController.abort(); + this->scope.close(); + } + + [[nodiscard]] bool isRunning() const override { + return this->running.load(); + } + +private: + void scheduleFindNeighbors(std::vector excluded) { + this->scope.add( + streamr::utils::co_withExecutor( + &this->executor, + folly::coro::co_invoke( + [this, excluded = std::move(excluded)]() mutable + -> folly::coro::Task { + co_await this->findNeighbors(std::move(excluded)); + }))); + } + + folly::coro::Task findNeighbors(std::vector excluded) { + if (!this->running.load()) { + co_return; + } + auto newExcludes = + co_await this->options.doFindNeighbors(std::move(excluded)); + std::set uniqueContacts; + for (const auto* view : + {&this->options.nearbyNodeView, + &this->options.leftNodeView, + &this->options.rightNodeView, + &this->options.randomNodeView}) { + for (const auto& id : view->getIds()) { + uniqueContacts.insert(id); + } + } + const auto uniqueContactCount = uniqueContacts.size(); + if (this->options.neighbors.size() < this->options.minCount && + newExcludes.size() < uniqueContactCount) { + AbortableTimers::setAbortableTimeout( + [this, newExcludes]() { + this->scheduleFindNeighbors(newExcludes); + }, + neighborFinderInterval, + this->abortController.getSignal()); + } else if ( + this->options.neighbors.size() == 0 && uniqueContactCount > 0) { + SLogger::debug( + "No neighbors found yet contacts are available, restarting " + "handshaking process"); + AbortableTimers::setAbortableTimeout( + [this]() { this->scheduleFindNeighbors({}); }, + neighborFinderInterval, + this->abortController.getSignal()); + } else { + this->running.store(false); + } + } +}; + +} // namespace streamr::trackerlessnetwork::neighbordiscovery diff --git a/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/NeighborUpdateManager.cppm b/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/NeighborUpdateManager.cppm new file mode 100644 index 00000000..d34d3aef --- /dev/null +++ b/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/NeighborUpdateManager.cppm @@ -0,0 +1,175 @@ +// Module streamr.trackerlessnetwork.NeighborUpdateManager +// Ported from packages/trackerless-network/src/content-delivery-layer/ +// neighbor-discovery/NeighborUpdateManager.ts (v103.8.0-rc.3): sends the +// local neighbor list to every neighbor at a fixed interval, records the +// round-trip time, and drops neighbors that ask to be removed. +// +// Adaptation: the TS `await scheduleAtInterval(...)` becomes a bounded +// task on a GuardedAsyncScope (each round is a set of RPCs with +// timeouts, and stop() aborts the interval before draining the scope). +module; + +// Coroutine definitions need std::coroutine_traits declared in THIS +// translation unit; it cannot arrive through an imported BMI. +#include // IWYU pragma: keep + +#include +#include +#include +#include +#include "packages/dht/protos/DhtRpc.pb.h" +#include "packages/network/protos/NetworkRpc.pb.h" + +export module streamr.trackerlessnetwork.NeighborUpdateManager; + +import streamr.utils.CoroutineHelper; +import streamr.utils.AbortController; +import streamr.utils.GuardedAsyncScope; +import streamr.utils.scheduleAtInterval; +import streamr.utils.SharedExecutors; +import streamr.trackerlessnetwork.NeighborFinder; +import streamr.trackerlessnetwork.NeighborUpdateRpcLocal; +import streamr.trackerlessnetwork.NeighborUpdateRpcRemote; +import streamr.trackerlessnetwork.NetworkRpcClient; +import streamr.trackerlessnetwork.NodeList; +import streamr.dht.DhtCallContext; +import streamr.dht.Identifiers; +import streamr.dht.ListeningRpcCommunicator; +import streamr.dht.protos; +import streamr.logger.SLogger; +import streamr.utils.StreamPartID; + +// Hoisted (file scope, NOT exported); fully qualified because relative +// namespace names resolve differently at file scope than inside the +// package namespace. +using streamr::dht::DhtAddress; +using streamr::dht::Identifiers; +using streamr::dht::rpcprotocol::DhtCallContext; +using streamr::dht::transport::ListeningRpcCommunicator; +using streamr::logger::SLogger; +using streamr::utils::AbortController; +using streamr::utils::GuardedAsyncScope; +using streamr::utils::scheduleAtInterval; +using streamr::utils::SharedSerialExecutor; +using streamr::utils::StreamPartID; + +export namespace streamr::trackerlessnetwork::neighbordiscovery { + +using ::dht::PeerDescriptor; + +constexpr std::chrono::milliseconds defaultNeighborUpdateInterval{10000}; + +struct NeighborUpdateManagerOptions { + PeerDescriptor localPeerDescriptor; + StreamPartID streamPartId; + NodeList& neighbors; + NodeList& nearbyNodeView; + INeighborFinder& neighborFinder; + ListeningRpcCommunicator& rpcCommunicator; + std::chrono::milliseconds neighborUpdateInterval; + size_t neighborTargetCount; + std::set& ongoingHandshakes; +}; + +class NeighborUpdateManager { +private: + NeighborUpdateManagerOptions options; + NeighborUpdateRpcLocal rpcLocal; + AbortController abortController; + SharedSerialExecutor executor{streamr::utils::SharedExecutors::worker()}; + GuardedAsyncScope scope; + +public: + explicit NeighborUpdateManager(NeighborUpdateManagerOptions options) + : options(options), // NOLINT(performance-unnecessary-value-param) + rpcLocal( + NeighborUpdateRpcLocalOptions{ + .localPeerDescriptor = options.localPeerDescriptor, + .streamPartId = options.streamPartId, + .neighbors = options.neighbors, + .nearbyNodeView = options.nearbyNodeView, + .neighborFinder = options.neighborFinder, + .rpcCommunicator = options.rpcCommunicator, + .neighborTargetCount = options.neighborTargetCount, + .ongoingHandshakes = options.ongoingHandshakes}) { + this->options.rpcCommunicator + .registerRpcMethod( + "neighborUpdate", + [this]( + const NeighborUpdate& request, + const DhtCallContext& context) { + return this->rpcLocal.neighborUpdate(request, context); + }); + } + + ~NeighborUpdateManager() { this->stop(); } + + void start() { + this->scope.add( + streamr::utils::co_withExecutor( + &this->executor, + folly::coro::co_invoke([this]() -> folly::coro::Task { + co_await scheduleAtInterval( + [this]() -> folly::coro::Task { + co_await this->updateNeighborInfo(); + }, + this->options.neighborUpdateInterval, + false, + this->abortController.getSignal(), + &this->executor); + }))); + } + + void stop() { + this->abortController.abort(); + this->scope.close(); + } + +private: + folly::coro::Task updateNeighborInfo() { + SLogger::trace("Updating neighbor info to nodes"); + std::vector neighborDescriptors; + for (const auto& neighbor : this->options.neighbors.getAll()) { + neighborDescriptors.push_back(neighbor->getPeerDescriptor()); + } + const auto startTime = std::chrono::steady_clock::now(); + std::vector> updates; + for (const auto& neighbor : this->options.neighbors.getAll()) { + updates.push_back(this->updateNeighbor( + neighbor->getPeerDescriptor(), neighborDescriptors, startTime)); + } + co_await folly::coro::collectAllTryRange(std::move(updates)); + } + + folly::coro::Task updateNeighbor( + PeerDescriptor peerDescriptor, + std::vector neighborDescriptors, + std::chrono::steady_clock::time_point startTime) { + const auto response = + co_await this->createRemote(peerDescriptor) + ->updateNeighbors( + this->options.streamPartId, std::move(neighborDescriptors)); + const auto nodeId = + Identifiers::getNodeIdFromPeerDescriptor(peerDescriptor); + const auto neighbor = this->options.neighbors.get(nodeId); + if (neighbor.has_value()) { + neighbor.value()->setRtt( + std::chrono::duration_cast( + std::chrono::steady_clock::now() - startTime) + .count()); + } + if (response.removeMe) { + this->options.neighbors.remove(nodeId); + this->options.neighborFinder.start({nodeId}); + } + } + + std::shared_ptr createRemote( + const PeerDescriptor& targetPeerDescriptor) { + NeighborUpdateRpcClient client{this->options.rpcCommunicator}; + return std::make_shared( + this->options.localPeerDescriptor, targetPeerDescriptor, client); + } +}; + +} // namespace streamr::trackerlessnetwork::neighbordiscovery diff --git a/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/NeighborUpdateRpcLocal.cppm b/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/NeighborUpdateRpcLocal.cppm new file mode 100644 index 00000000..44803de4 --- /dev/null +++ b/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/NeighborUpdateRpcLocal.cppm @@ -0,0 +1,121 @@ +// Module streamr.trackerlessnetwork.NeighborUpdateRpcLocal +// Ported from packages/trackerless-network/src/content-delivery-layer/ +// neighbor-discovery/NeighborUpdateRpcLocal.ts (v103.8.0-rc.3): the +// server side of periodic neighbor-list exchange — learns new contacts +// from the caller's neighbor list and asks to be dropped when both sides +// have more neighbors than the target count. +module; + +#include +#include +#include +#include +#include "packages/dht/protos/DhtRpc.pb.h" +#include "packages/network/protos/NetworkRpc.pb.h" + +export module streamr.trackerlessnetwork.NeighborUpdateRpcLocal; + +import streamr.trackerlessnetwork.ContentDeliveryRpcRemote; +import streamr.trackerlessnetwork.NeighborFinder; +import streamr.trackerlessnetwork.NetworkRpcClient; +import streamr.trackerlessnetwork.NodeList; +import streamr.dht.DhtCallContext; +import streamr.dht.Identifiers; +import streamr.dht.ListeningRpcCommunicator; +import streamr.dht.protos; +import streamr.utils.StreamPartID; + +// Hoisted (file scope, NOT exported); fully qualified because relative +// namespace names resolve differently at file scope than inside the +// package namespace. +using streamr::dht::DhtAddress; +using streamr::dht::Identifiers; +using streamr::dht::rpcprotocol::DhtCallContext; +using streamr::dht::transport::ListeningRpcCommunicator; +using streamr::trackerlessnetwork::ContentDeliveryRpcClient; +using streamr::trackerlessnetwork::ContentDeliveryRpcRemote; +using streamr::utils::StreamPartID; + +export namespace streamr::trackerlessnetwork::neighbordiscovery { + +using ::dht::PeerDescriptor; + +struct NeighborUpdateRpcLocalOptions { + PeerDescriptor localPeerDescriptor; + StreamPartID streamPartId; + NodeList& neighbors; + NodeList& nearbyNodeView; + INeighborFinder& neighborFinder; + ListeningRpcCommunicator& rpcCommunicator; + size_t neighborTargetCount; + std::set& ongoingHandshakes; +}; + +class NeighborUpdateRpcLocal { +private: + NeighborUpdateRpcLocalOptions options; + +public: + explicit NeighborUpdateRpcLocal(NeighborUpdateRpcLocalOptions options) + : options(std::move(options)) {} + + NeighborUpdate neighborUpdate( + const NeighborUpdate& message, const DhtCallContext& context) { + const auto senderPeerDescriptor = + context.incomingSourceDescriptor.value(); + const auto remoteNodeId = + Identifiers::getNodeIdFromPeerDescriptor(senderPeerDescriptor); + this->updateContacts(message.neighbordescriptors()); + if (!this->options.neighbors.has(remoteNodeId) && + !this->options.ongoingHandshakes.contains(remoteNodeId)) { + return this->createResponse(true); + } + const auto isOverNeighborCount = this->options.neighbors.size() > + this->options.neighborTargetCount && + // Motivation: We don't know the remote's neighborTargetCount + // setting here. We only ask to cut connections if the remote + // has a "sufficient" number of neighbors, where "sufficient" + // means our neighborTargetCount setting. + static_cast(message.neighbordescriptors().size()) > + this->options.neighborTargetCount; + if (!isOverNeighborCount) { + this->options.neighborFinder.start(); + } else { + this->options.neighbors.remove(remoteNodeId); + } + return this->createResponse(isOverNeighborCount); + } + +private: + template + void updateContacts(const DescriptorRange& neighborDescriptors) { + const auto ownNodeId = Identifiers::getNodeIdFromPeerDescriptor( + this->options.localPeerDescriptor); + const auto knownIds = this->options.neighbors.getIds(); + for (const auto& peerDescriptor : neighborDescriptors) { + const auto nodeId = + Identifiers::getNodeIdFromPeerDescriptor(peerDescriptor); + if (nodeId != ownNodeId && + std::ranges::find(knownIds, nodeId) == knownIds.end()) { + ContentDeliveryRpcClient client{this->options.rpcCommunicator}; + this->options.nearbyNodeView.add( + std::make_shared( + this->options.localPeerDescriptor, + peerDescriptor, + client)); + } + } + } + + NeighborUpdate createResponse(bool removeMe) const { + NeighborUpdate response; + response.set_streampartid(this->options.streamPartId); + for (const auto& neighbor : this->options.neighbors.getAll()) { + *response.add_neighbordescriptors() = neighbor->getPeerDescriptor(); + } + response.set_removeme(removeMe); + return response; + } +}; + +} // namespace streamr::trackerlessnetwork::neighbordiscovery diff --git a/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/NeighborUpdateRpcRemote.cppm b/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/NeighborUpdateRpcRemote.cppm new file mode 100644 index 00000000..8af4f708 --- /dev/null +++ b/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/NeighborUpdateRpcRemote.cppm @@ -0,0 +1,90 @@ +// Module streamr.trackerlessnetwork.NeighborUpdateRpcRemote +// Ported from packages/trackerless-network/src/content-delivery-layer/ +// neighbor-discovery/NeighborUpdateRpcRemote.ts (v103.8.0-rc.3): the +// client side of periodic neighbor-list exchange. Errors are swallowed +// and reported as removeMe=true, matching the TS behavior. +module; + +// Coroutine definitions need std::coroutine_traits declared in THIS +// translation unit; it cannot arrive through an imported BMI. +#include // IWYU pragma: keep + +#include +#include +#include "packages/dht/protos/DhtRpc.pb.h" +#include "packages/network/protos/NetworkRpc.pb.h" + +export module streamr.trackerlessnetwork.NeighborUpdateRpcRemote; + +import streamr.utils.CoroutineHelper; +import streamr.trackerlessnetwork.NetworkRpcClient; +import streamr.dht.DhtCallContext; +import streamr.dht.Identifiers; +import streamr.dht.RpcRemote; +import streamr.dht.protos; +import streamr.logger.SLogger; +import streamr.utils.StreamPartID; + +// Hoisted (file scope, NOT exported); fully qualified because relative +// namespace names resolve differently at file scope than inside the +// package namespace. +using streamr::dht::Identifiers; +using streamr::dht::contact::RpcRemote; +using streamr::dht::rpcprotocol::DhtCallContext; +using streamr::logger::SLogger; +using streamr::utils::StreamPartID; + +export namespace streamr::trackerlessnetwork::neighbordiscovery { + +using ::dht::PeerDescriptor; +using NeighborUpdateRpcClient = + streamr::protorpc::NeighborUpdateRpcClient; + +struct UpdateNeighborsResponse { + std::vector peerDescriptors; + bool removeMe = false; +}; + +class NeighborUpdateRpcRemote : public RpcRemote { +public: + NeighborUpdateRpcRemote( + PeerDescriptor localPeerDescriptor, // NOLINT + PeerDescriptor remotePeerDescriptor, + NeighborUpdateRpcClient client, + std::optional timeout = std::nullopt) + : RpcRemote( + std::move(localPeerDescriptor), + std::move(remotePeerDescriptor), + client, + timeout) {} + + folly::coro::Task updateNeighbors( + StreamPartID streamPartId, std::vector neighbors) { + NeighborUpdate request; + request.set_streampartid(streamPartId); + for (auto& neighbor : neighbors) { + *request.add_neighbordescriptors() = std::move(neighbor); + } + request.set_removeme(false); + auto options = this->formDhtRpcOptions({}); + try { + const auto response = co_await this->getClient().neighborUpdate( + std::move(request), std::move(options)); + UpdateNeighborsResponse result; + result.peerDescriptors.assign( + response.neighbordescriptors().begin(), + response.neighbordescriptors().end()); + result.removeMe = response.removeme(); + co_return result; + } catch (const std::exception& err) { + SLogger::debug( + "updateNeighbors to " + + Identifiers::getNodeIdFromPeerDescriptor( + this->getPeerDescriptor()) + + " failed: " + std::string(err.what())); + co_return UpdateNeighborsResponse{.removeMe = true}; + } + } +}; + +} // namespace streamr::trackerlessnetwork::neighbordiscovery diff --git a/packages/streamr-trackerless-network/test/unit/HandshakeRpcLocalTest.cpp b/packages/streamr-trackerless-network/test/unit/HandshakeRpcLocalTest.cpp new file mode 100644 index 00000000..6f485146 --- /dev/null +++ b/packages/streamr-trackerless-network/test/unit/HandshakeRpcLocalTest.cpp @@ -0,0 +1,178 @@ +// Ported from packages/trackerless-network/test/unit/ +// HandshakeRpcLocal.test.ts (v103.8.0-rc.3): accept/reject decisions of +// the handshake server, including the interleaving branches. +// +// NB: NetworkRpc types are consumed ONLY through the +// streamr.trackerlessnetwork.protos module (no textual NetworkRpc.pb.h +// include) — see TestUtilsTest.cpp for the clangd rationale. +#include +#include +#include +#include +#include "packages/dht/protos/DhtRpc.pb.h" + +#include // IWYU pragma: keep + +import streamr.utils.CoroutineHelper; +import streamr.trackerlessnetwork.HandshakeRpcLocal; +import streamr.trackerlessnetwork.NodeList; +import streamr.trackerlessnetwork.TestUtils; +import streamr.trackerlessnetwork.protos; +import streamr.dht.DhtCallContext; +import streamr.dht.Identifiers; +import streamr.dht.protos; +import streamr.utils.StreamPartID; +import streamr.utils.waitForCondition; + +using ::dht::PeerDescriptor; +using streamr::dht::DhtAddress; +using streamr::dht::Identifiers; +using streamr::dht::rpcprotocol::DhtCallContext; +using streamr::trackerlessnetwork::NodeList; +using streamr::trackerlessnetwork::neighbordiscovery::HandshakeRpcLocal; +using streamr::trackerlessnetwork::neighbordiscovery::HandshakeRpcLocalOptions; +using streamr::trackerlessnetwork::testutils:: + createMockContentDeliveryRpcRemote; +using streamr::trackerlessnetwork::testutils::createMockHandshakeRpcRemote; +using streamr::trackerlessnetwork::testutils::createMockPeerDescriptor; +using streamr::utils::blockingWait; +using streamr::utils::StreamPartIDUtils; + +namespace { +constexpr size_t neighborsLimit = 10; +} // namespace + +class HandshakeRpcLocalTest : public ::testing::Test { +protected: + PeerDescriptor localPeerDescriptor = createMockPeerDescriptor(); + NodeList neighbors{ + Identifiers::getNodeIdFromPeerDescriptor(localPeerDescriptor), + neighborsLimit}; + std::set ongoingHandshakes; + std::set ongoingInterleaves; + size_t handshakeWithInterleavingCalls = 0; + std::optional rpcLocal; + + void SetUp() override { + this->rpcLocal.emplace( + HandshakeRpcLocalOptions{ + .streamPartId = StreamPartIDUtils::parse("stream#0"), + .neighbors = this->neighbors, + .ongoingHandshakes = this->ongoingHandshakes, + .ongoingInterleaves = this->ongoingInterleaves, + .maxNeighborCount = 4, + .createRpcRemote = + [](const PeerDescriptor& /*target*/) { + return createMockHandshakeRpcRemote(); + }, + .createContentDeliveryRpcRemote = + [](const PeerDescriptor& /*target*/) { + return createMockContentDeliveryRpcRemote(); + }, + .handshakeWithInterleaving = [this]( + PeerDescriptor /*target*/, + DhtAddress /*remoteNodeId*/) + -> folly::coro::Task { + this->handshakeWithInterleavingCalls++; + co_return true; + }}); + } + + static StreamPartHandshakeRequest createRequest() { + StreamPartHandshakeRequest request; + request.set_streampartid(StreamPartIDUtils::parse("stream#0")); + request.set_requestid("requestId"); + return request; + } + + static DhtCallContext createContext(const PeerDescriptor& sender) { + DhtCallContext context; + context.incomingSourceDescriptor = sender; + return context; + } +}; + +TEST_F(HandshakeRpcLocalTest, Handshake) { + const auto response = this->rpcLocal->handshake( + createRequest(), createContext(createMockPeerDescriptor())); + EXPECT_EQ(response.accepted(), true); + EXPECT_EQ(response.has_interleavetargetdescriptor(), false); + EXPECT_EQ(response.requestid(), "requestId"); +} + +TEST_F(HandshakeRpcLocalTest, HandshakeInterleave) { + this->neighbors.add(createMockContentDeliveryRpcRemote()); + this->neighbors.add(createMockContentDeliveryRpcRemote()); + this->neighbors.add(createMockContentDeliveryRpcRemote()); + this->neighbors.add(createMockContentDeliveryRpcRemote()); + const auto response = this->rpcLocal->handshake( + createRequest(), createContext(createMockPeerDescriptor())); + EXPECT_EQ(response.accepted(), true); + EXPECT_EQ(response.has_interleavetargetdescriptor(), true); +} + +TEST_F(HandshakeRpcLocalTest, UnacceptedHandshake) { + this->ongoingHandshakes.insert(DhtAddress{"0x2222"}); + this->ongoingHandshakes.insert(DhtAddress{"0x3333"}); + this->ongoingHandshakes.insert(DhtAddress{"0x4444"}); + this->ongoingHandshakes.insert(DhtAddress{"0x5555"}); + const auto response = this->rpcLocal->handshake( + createRequest(), createContext(createMockPeerDescriptor())); + EXPECT_EQ(response.accepted(), false); +} + +TEST_F(HandshakeRpcLocalTest, HandshakeWithInterleavingSuccess) { + InterleaveRequest request; + *request.mutable_interleavetargetdescriptor() = createMockPeerDescriptor(); + const auto response = blockingWait(this->rpcLocal->interleaveRequest( + request, createContext(createMockPeerDescriptor()))); + EXPECT_EQ(response.accepted(), true); + EXPECT_EQ(this->handshakeWithInterleavingCalls, 1); +} + +TEST_F( + HandshakeRpcLocalTest, + RejectsHandshakesIfInterleavingToRequestorIsOngoing) { + this->neighbors.add(createMockContentDeliveryRpcRemote()); + this->neighbors.add(createMockContentDeliveryRpcRemote()); + this->neighbors.add(createMockContentDeliveryRpcRemote()); + this->neighbors.add(createMockContentDeliveryRpcRemote()); + const auto requestor = createMockPeerDescriptor(); + this->ongoingInterleaves.insert( + Identifiers::getNodeIdFromPeerDescriptor(requestor)); + const auto response = + this->rpcLocal->handshake(createRequest(), createContext(requestor)); + EXPECT_EQ(response.accepted(), false); +} + +TEST_F(HandshakeRpcLocalTest, RejectsIfTooManyInterleavingRequestsOngoing) { + const auto interleavingPeer1 = createMockPeerDescriptor(); + const auto interleavingPeer2 = createMockPeerDescriptor(); + const auto interleavingPeer3 = createMockPeerDescriptor(); + this->neighbors.add(createMockContentDeliveryRpcRemote(interleavingPeer1)); + this->neighbors.add(createMockContentDeliveryRpcRemote(interleavingPeer2)); + this->neighbors.add(createMockContentDeliveryRpcRemote(interleavingPeer3)); + this->neighbors.add(createMockContentDeliveryRpcRemote()); + this->ongoingInterleaves.insert( + Identifiers::getNodeIdFromPeerDescriptor(interleavingPeer1)); + this->ongoingInterleaves.insert( + Identifiers::getNodeIdFromPeerDescriptor(interleavingPeer2)); + this->ongoingInterleaves.insert( + Identifiers::getNodeIdFromPeerDescriptor(interleavingPeer3)); + const auto response = this->rpcLocal->handshake( + createRequest(), createContext(createMockPeerDescriptor())); + EXPECT_EQ(response.accepted(), false); + EXPECT_EQ(this->handshakeWithInterleavingCalls, 0); +} + +TEST_F(HandshakeRpcLocalTest, RejectsIfRequestorHasMoreThanMaxNeighborCount) { + this->neighbors.add(createMockContentDeliveryRpcRemote()); + this->neighbors.add(createMockContentDeliveryRpcRemote()); + this->neighbors.add(createMockContentDeliveryRpcRemote()); + this->neighbors.add(createMockContentDeliveryRpcRemote()); + this->neighbors.add(createMockContentDeliveryRpcRemote()); + const auto response = this->rpcLocal->handshake( + createRequest(), createContext(createMockPeerDescriptor())); + EXPECT_EQ(response.accepted(), false); + EXPECT_EQ(this->handshakeWithInterleavingCalls, 0); +} diff --git a/packages/streamr-trackerless-network/test/unit/HandshakeRpcRemoteTest.cpp b/packages/streamr-trackerless-network/test/unit/HandshakeRpcRemoteTest.cpp new file mode 100644 index 00000000..a7956202 --- /dev/null +++ b/packages/streamr-trackerless-network/test/unit/HandshakeRpcRemoteTest.cpp @@ -0,0 +1,86 @@ +// Ported from packages/trackerless-network/test/integration/ +// HandshakeRpcRemote.test.ts (v103.8.0-rc.3). Adaptation: the house +// two-communicator pattern instead of simulator transports (see +// ContentDeliveryRpcRemoteTest.cpp). +// +// NB: NetworkRpc types are consumed ONLY through the +// streamr.trackerlessnetwork.protos module (no textual NetworkRpc.pb.h +// include) — see TestUtilsTest.cpp for the clangd rationale. +#include +#include +#include +#include "packages/dht/protos/DhtRpc.pb.h" + +#include // IWYU pragma: keep + +import streamr.utils.CoroutineHelper; +import streamr.protorpc.RpcCommunicator; +import streamr.protorpc.protos; +import streamr.trackerlessnetwork.HandshakeRpcRemote; +import streamr.trackerlessnetwork.NetworkRpcClient; +import streamr.trackerlessnetwork.TestUtils; +import streamr.trackerlessnetwork.protos; +import streamr.dht.DhtCallContext; +import streamr.dht.protos; +import streamr.utils.StreamPartID; + +using ::dht::PeerDescriptor; +using ::protorpc::RpcMessage; +using streamr::dht::rpcprotocol::DhtCallContext; +using streamr::protorpc::RpcCommunicator; +using streamr::trackerlessnetwork::neighbordiscovery::HandshakeRpcClient; +using streamr::trackerlessnetwork::neighbordiscovery::HandshakeRpcRemote; +using streamr::trackerlessnetwork::testutils::createMockPeerDescriptor; +using streamr::utils::blockingWait; +using streamr::utils::StreamPartIDUtils; + +using RpcCommunicatorType = RpcCommunicator; + +class HandshakeRpcRemoteTest : public ::testing::Test { +protected: + RpcCommunicatorType clientCommunicator; + RpcCommunicatorType serverCommunicator; + PeerDescriptor clientNode = createMockPeerDescriptor(); + PeerDescriptor serverNode = createMockPeerDescriptor(); + std::optional rpcRemote; + + void SetUp() override { + this->serverCommunicator.registerRpcMethod< + StreamPartHandshakeRequest, + StreamPartHandshakeResponse>( + "handshake", + [](const StreamPartHandshakeRequest& request, + const DhtCallContext& /*context*/) { + StreamPartHandshakeResponse response; + response.set_requestid(request.requestid()); + response.set_accepted(true); + return response; + }); + this->clientCommunicator.setOutgoingMessageCallback( + [this]( + const RpcMessage& message, + const std::string& /*requestId*/, + const DhtCallContext& /*context*/) { + this->serverCommunicator.handleIncomingMessage( + message, DhtCallContext()); + }); + this->serverCommunicator.setOutgoingMessageCallback( + [this]( + const RpcMessage& message, + const std::string& /*requestId*/, + const DhtCallContext& /*context*/) { + this->clientCommunicator.handleIncomingMessage( + message, DhtCallContext()); + }); + this->rpcRemote.emplace( + this->clientNode, + this->serverNode, + HandshakeRpcClient(this->clientCommunicator)); + } +}; + +TEST_F(HandshakeRpcRemoteTest, Handshake) { + const auto result = blockingWait( + this->rpcRemote->handshake(StreamPartIDUtils::parse("test#0"), {})); + EXPECT_EQ(result.accepted, true); +} diff --git a/packages/streamr-trackerless-network/test/unit/HandshakerTest.cpp b/packages/streamr-trackerless-network/test/unit/HandshakerTest.cpp new file mode 100644 index 00000000..b284213e --- /dev/null +++ b/packages/streamr-trackerless-network/test/unit/HandshakerTest.cpp @@ -0,0 +1,96 @@ +// Ported from packages/trackerless-network/test/unit/Handshaker.test.ts +// (v103.8.0-rc.3): target selection with empty views and with +// unreachable contacts. Adaptation: the TS test runs over +// Simulator/SimulatorTransport; a FakeTransport whose send callback +// throws gives the same observable behavior (handshakes to the mock +// contacts fail) without waiting for RPC timeouts. +#include +#include +#include +#include +#include +#include "packages/dht/protos/DhtRpc.pb.h" + +#include // IWYU pragma: keep + +import streamr.utils.CoroutineHelper; +import streamr.trackerlessnetwork.Handshaker; +import streamr.trackerlessnetwork.NodeList; +import streamr.trackerlessnetwork.TestUtils; +import streamr.dht.DhtCallContext; +import streamr.dht.FakeTransport; +import streamr.dht.Identifiers; +import streamr.dht.ListeningRpcCommunicator; +import streamr.dht.protos; +import streamr.utils.StreamPartID; + +using ::dht::PeerDescriptor; +using streamr::dht::DhtAddress; +using streamr::dht::Identifiers; +using streamr::dht::ServiceID; +using streamr::dht::transport::FakeTransport; +using streamr::dht::transport::ListeningRpcCommunicator; +using streamr::trackerlessnetwork::NodeList; +using streamr::trackerlessnetwork::neighbordiscovery::Handshaker; +using streamr::trackerlessnetwork::neighbordiscovery::HandshakerOptions; +using streamr::trackerlessnetwork::testutils:: + createMockContentDeliveryRpcRemote; +using streamr::trackerlessnetwork::testutils::createMockPeerDescriptor; +using streamr::utils::blockingWait; +using streamr::utils::StreamPartIDUtils; + +namespace { +constexpr size_t maxNeighborCount = 4; +constexpr size_t neighborsLimit = 10; +constexpr size_t viewLimit = 20; +constexpr std::chrono::milliseconds rpcRequestTimeout{5000}; +} // namespace + +class HandshakerTest : public ::testing::Test { +protected: + PeerDescriptor peerDescriptor = createMockPeerDescriptor(); + FakeTransport transport{peerDescriptor, [](const auto& /*message*/) { + throw std::runtime_error("unreachable"); + }}; + ListeningRpcCommunicator rpcCommunicator{ + ServiceID{StreamPartIDUtils::parse("stream#0")}, transport}; + DhtAddress nodeId = + Identifiers::getNodeIdFromPeerDescriptor(this->peerDescriptor); + NodeList neighbors{nodeId, neighborsLimit}; + NodeList leftNodeView{nodeId, viewLimit}; + NodeList rightNodeView{nodeId, viewLimit}; + NodeList nearbyNodeView{nodeId, viewLimit}; + NodeList randomNodeView{nodeId, viewLimit}; + std::set ongoingHandshakes; + std::optional handshaker; + + void SetUp() override { + this->handshaker.emplace( + HandshakerOptions{ + .localPeerDescriptor = this->peerDescriptor, + .streamPartId = StreamPartIDUtils::parse("stream#0"), + .neighbors = this->neighbors, + .leftNodeView = this->leftNodeView, + .rightNodeView = this->rightNodeView, + .nearbyNodeView = this->nearbyNodeView, + .randomNodeView = this->randomNodeView, + .rpcCommunicator = this->rpcCommunicator, + .maxNeighborCount = maxNeighborCount, + .ongoingHandshakes = this->ongoingHandshakes, + .rpcRequestTimeout = rpcRequestTimeout}); + } +}; + +TEST_F(HandshakerTest, AttemptHandshakesOnContactWorksWithEmptyStructures) { + const auto result = + blockingWait(this->handshaker->attemptHandshakesOnContacts({})); + EXPECT_EQ(result.size(), 0); +} + +TEST_F(HandshakerTest, AttemptHandshakesWithKnownNodesThatCannotBeConnectedTo) { + this->randomNodeView.add(createMockContentDeliveryRpcRemote()); + this->randomNodeView.add(createMockContentDeliveryRpcRemote()); + const auto result = + blockingWait(this->handshaker->attemptHandshakesOnContacts({})); + EXPECT_EQ(result.size(), 2); +} diff --git a/packages/streamr-trackerless-network/test/unit/HandshakesTest.cpp b/packages/streamr-trackerless-network/test/unit/HandshakesTest.cpp new file mode 100644 index 00000000..2e9a3088 --- /dev/null +++ b/packages/streamr-trackerless-network/test/unit/HandshakesTest.cpp @@ -0,0 +1,205 @@ +// Ported from packages/trackerless-network/test/integration/ +// Handshakes.test.ts (v103.8.0-rc.3): a real Handshaker (node 2) shakes +// hands with node 1 over simulator transports; node 1 accepts, rejects, +// or redirects to node 3 via the interleaving protocol. +// +// NB: NetworkRpc types are consumed ONLY through the +// streamr.trackerlessnetwork.protos module (no textual NetworkRpc.pb.h +// include) — see TestUtilsTest.cpp for the clangd rationale. +#include +#include +#include +#include +#include "packages/dht/protos/DhtRpc.pb.h" + +#include // IWYU pragma: keep + +import streamr.utils.CoroutineHelper; +import streamr.trackerlessnetwork.ContentDeliveryRpcRemote; +import streamr.trackerlessnetwork.Handshaker; +import streamr.trackerlessnetwork.NetworkRpcClient; +import streamr.trackerlessnetwork.NodeList; +import streamr.trackerlessnetwork.TestUtils; +import streamr.trackerlessnetwork.protos; +import streamr.dht.DhtCallContext; +import streamr.dht.Identifiers; +import streamr.dht.ListeningRpcCommunicator; +import streamr.dht.Simulator; +import streamr.dht.SimulatorTransport; +import streamr.dht.protos; +import streamr.utils.StreamPartID; + +using ::dht::PeerDescriptor; +using streamr::dht::DhtAddress; +using streamr::dht::Identifiers; +using streamr::dht::ServiceID; +using streamr::dht::connection::simulator::LatencyType; +using streamr::dht::connection::simulator::Simulator; +using streamr::dht::connection::simulator::SimulatorTransport; +using streamr::dht::rpcprotocol::DhtCallContext; +using streamr::dht::transport::ListeningRpcCommunicator; +using streamr::trackerlessnetwork::ContentDeliveryRpcClient; +using streamr::trackerlessnetwork::ContentDeliveryRpcRemote; +using streamr::trackerlessnetwork::NodeList; +using streamr::trackerlessnetwork::neighbordiscovery::Handshaker; +using streamr::trackerlessnetwork::neighbordiscovery::HandshakerOptions; +using streamr::trackerlessnetwork::testutils::createMockPeerDescriptor; +using streamr::utils::blockingWait; +using streamr::utils::StreamPartIDUtils; + +namespace { +constexpr size_t viewLimit = 10; +constexpr size_t maxNeighborCount = 4; +} // namespace + +class HandshakesTest : public ::testing::Test { +protected: + PeerDescriptor peerDescriptor1 = createMockPeerDescriptor(); + PeerDescriptor peerDescriptor2 = createMockPeerDescriptor(); + PeerDescriptor peerDescriptor3 = createMockPeerDescriptor(); + + Simulator simulator{LatencyType::NONE}; + std::shared_ptr simulatorTransport1; + std::shared_ptr simulatorTransport2; + std::shared_ptr simulatorTransport3; + std::shared_ptr rpcCommunicator1; + std::shared_ptr rpcCommunicator2; + std::shared_ptr rpcCommunicator3; + + std::optional neighbors; + std::optional leftNodeView; + std::optional rightNodeView; + std::optional nodeView; + std::set ongoingHandshakes; + std::optional handshaker; + + void SetUp() override { + const auto streamPartId = StreamPartIDUtils::parse("stream#0"); + this->simulatorTransport1 = std::make_shared( + this->peerDescriptor1, this->simulator); + this->simulatorTransport1->start(); + this->simulatorTransport2 = std::make_shared( + this->peerDescriptor2, this->simulator); + this->simulatorTransport2->start(); + this->simulatorTransport3 = std::make_shared( + this->peerDescriptor3, this->simulator); + this->simulatorTransport3->start(); + this->rpcCommunicator1 = std::make_shared( + ServiceID{streamPartId}, *this->simulatorTransport1); + this->rpcCommunicator2 = std::make_shared( + ServiceID{streamPartId}, *this->simulatorTransport2); + this->rpcCommunicator3 = std::make_shared( + ServiceID{streamPartId}, *this->simulatorTransport3); + + const auto handshakerNodeId = + Identifiers::getNodeIdFromPeerDescriptor(this->peerDescriptor2); + this->leftNodeView.emplace(handshakerNodeId, viewLimit); + this->rightNodeView.emplace(handshakerNodeId, viewLimit); + this->nodeView.emplace(handshakerNodeId, viewLimit); + ContentDeliveryRpcClient client{*this->rpcCommunicator2}; + this->nodeView->add( + std::make_shared( + this->peerDescriptor2, this->peerDescriptor1, client)); + this->neighbors.emplace(handshakerNodeId, maxNeighborCount); + this->handshaker.emplace( + HandshakerOptions{ + .localPeerDescriptor = this->peerDescriptor2, + .streamPartId = streamPartId, + .neighbors = this->neighbors.value(), + .leftNodeView = this->leftNodeView.value(), + .rightNodeView = this->rightNodeView.value(), + .nearbyNodeView = this->nodeView.value(), + .randomNodeView = this->nodeView.value(), + .rpcCommunicator = *this->rpcCommunicator2, + .maxNeighborCount = maxNeighborCount, + .ongoingHandshakes = this->ongoingHandshakes}); + } + + void TearDown() override { + // The C++ ListeningRpcCommunicator has no explicit stop(); + // destruction (after the transports stop) drains it. + this->simulatorTransport1->stop(); + this->simulatorTransport2->stop(); + this->simulatorTransport3->stop(); + this->simulator.stop(); + } + + static void registerAcceptingHandshakeHandler( + ListeningRpcCommunicator& communicator) { + communicator.registerRpcMethod< + StreamPartHandshakeRequest, + StreamPartHandshakeResponse>( + "handshake", + [](const StreamPartHandshakeRequest& request, + const DhtCallContext& /*context*/) { + StreamPartHandshakeResponse response; + response.set_requestid(request.requestid()); + response.set_accepted(true); + return response; + }); + } +}; + +TEST_F(HandshakesTest, HandshakeAccepted) { + registerAcceptingHandshakeHandler(*this->rpcCommunicator1); + const auto result = + blockingWait(this->handshaker->attemptHandshakesOnContacts({})); + EXPECT_EQ(result.size(), 0); + EXPECT_EQ( + this->neighbors->has( + Identifiers::getNodeIdFromPeerDescriptor(this->peerDescriptor1)), + true); +} + +TEST_F(HandshakesTest, HandshakeRejected) { + this->rpcCommunicator1->registerRpcMethod< + StreamPartHandshakeRequest, + StreamPartHandshakeResponse>( + "handshake", + [](const StreamPartHandshakeRequest& request, + const DhtCallContext& /*context*/) { + StreamPartHandshakeResponse response; + response.set_requestid(request.requestid()); + response.set_accepted(false); + return response; + }); + const auto result = + blockingWait(this->handshaker->attemptHandshakesOnContacts({})); + ASSERT_EQ(result.size(), 1); + EXPECT_EQ( + result[0], + Identifiers::getNodeIdFromPeerDescriptor(this->peerDescriptor1)); + EXPECT_EQ( + this->neighbors->has( + Identifiers::getNodeIdFromPeerDescriptor(this->peerDescriptor1)), + false); +} + +TEST_F(HandshakesTest, HandshakeWithInterleaving) { + this->rpcCommunicator1->registerRpcMethod< + StreamPartHandshakeRequest, + StreamPartHandshakeResponse>( + "handshake", + [this]( + const StreamPartHandshakeRequest& request, + const DhtCallContext& /*context*/) { + StreamPartHandshakeResponse response; + response.set_requestid(request.requestid()); + response.set_accepted(true); + *response.mutable_interleavetargetdescriptor() = + this->peerDescriptor3; + return response; + }); + registerAcceptingHandshakeHandler(*this->rpcCommunicator3); + const auto result = + blockingWait(this->handshaker->attemptHandshakesOnContacts({})); + EXPECT_EQ(result.size(), 0); + EXPECT_EQ( + this->neighbors->has( + Identifiers::getNodeIdFromPeerDescriptor(this->peerDescriptor1)), + true); + EXPECT_EQ( + this->neighbors->has( + Identifiers::getNodeIdFromPeerDescriptor(this->peerDescriptor3)), + true); +} diff --git a/packages/streamr-trackerless-network/test/unit/NeighborFinderTest.cpp b/packages/streamr-trackerless-network/test/unit/NeighborFinderTest.cpp new file mode 100644 index 00000000..2022aff8 --- /dev/null +++ b/packages/streamr-trackerless-network/test/unit/NeighborFinderTest.cpp @@ -0,0 +1,93 @@ +// Ported from packages/trackerless-network/test/unit/ +// NeighborFinder.test.ts (v103.8.0-rc.3): the finder keeps running +// rounds until the neighbor list reaches minCount, then stops itself. +#include +#include +#include +#include +#include + +#include // IWYU pragma: keep + +import streamr.utils.CoroutineHelper; +import streamr.trackerlessnetwork.NeighborFinder; +import streamr.trackerlessnetwork.NodeList; +import streamr.trackerlessnetwork.TestUtils; +import streamr.dht.Identifiers; +import streamr.utils.waitForCondition; + +using streamr::dht::DhtAddress; +using streamr::dht::Identifiers; +using streamr::trackerlessnetwork::NodeList; +using streamr::trackerlessnetwork::neighbordiscovery::NeighborFinder; +using streamr::trackerlessnetwork::neighbordiscovery::NeighborFinderOptions; +using streamr::trackerlessnetwork::testutils:: + createMockContentDeliveryRpcRemote; +using streamr::utils::blockingWait; +using streamr::utils::waitForCondition; + +namespace { +constexpr size_t minCount = 4; +constexpr size_t nearbyContactCount = 30; +constexpr size_t neighborsLimit = 15; +constexpr size_t viewLimit = 30; +constexpr std::chrono::seconds findTimeout{10}; +constexpr std::chrono::seconds stopTimeout{5}; +constexpr std::chrono::milliseconds findPollInterval{100}; +constexpr std::chrono::milliseconds stopPollInterval{50}; +} // namespace + +class NeighborFinderTest : public ::testing::Test { +protected: + DhtAddress nodeId = Identifiers::createRandomDhtAddress(); + NodeList neighbors{nodeId, neighborsLimit}; + NodeList nearbyNodeView{nodeId, nearbyContactCount}; + NodeList leftNodeView{nodeId, viewLimit}; + NodeList rightNodeView{nodeId, viewLimit}; + NodeList randomNodeView{nodeId, viewLimit}; + std::optional neighborFinder; + + void SetUp() override { + for (size_t i = 0; i < nearbyContactCount; i++) { + this->nearbyNodeView.add(createMockContentDeliveryRpcRemote()); + } + this->neighborFinder.emplace( + NeighborFinderOptions{ + .neighbors = this->neighbors, + .nearbyNodeView = this->nearbyNodeView, + .leftNodeView = this->leftNodeView, + .rightNodeView = this->rightNodeView, + .randomNodeView = this->randomNodeView, + .doFindNeighbors = [this](std::vector excluded) + -> folly::coro::Task> { + const auto target = + this->nearbyNodeView.getRandom(excluded); + if (rand() % 2 == 0) { // NOLINT + this->neighbors.add(target.value()); + } else { + excluded.push_back( + Identifiers::getNodeIdFromPeerDescriptor( + target.value()->getPeerDescriptor())); + } + co_return excluded; + }, + .minCount = minCount}); + } + + void TearDown() override { this->neighborFinder->stop(); } +}; + +TEST_F(NeighborFinderTest, FindsTargetNumberOfNodes) { + this->neighborFinder->start(); + blockingWait(waitForCondition( + [this]() { return this->neighbors.size() >= minCount; }, + findTimeout, + findPollInterval)); + // The finder marks itself stopped at the end of the round that + // reached the target, so allow that round to finish. + blockingWait(waitForCondition( + [this]() { return !this->neighborFinder->isRunning(); }, + stopTimeout, + stopPollInterval)); + EXPECT_EQ(this->neighborFinder->isRunning(), false); +} diff --git a/packages/streamr-trackerless-network/test/unit/NeighborUpdateRpcLocalTest.cpp b/packages/streamr-trackerless-network/test/unit/NeighborUpdateRpcLocalTest.cpp new file mode 100644 index 00000000..beb61a02 --- /dev/null +++ b/packages/streamr-trackerless-network/test/unit/NeighborUpdateRpcLocalTest.cpp @@ -0,0 +1,194 @@ +// Ported from packages/trackerless-network/test/unit/ +// NeighborUpdateRpcLocal.test.ts (v103.8.0-rc.3): the neighbor-update +// server learns contacts from the caller and asks to be removed only in +// the right situations. +// +// NB: NetworkRpc types are consumed ONLY through the +// streamr.trackerlessnetwork.protos module (no textual NetworkRpc.pb.h +// include) — see TestUtilsTest.cpp for the clangd rationale. +#include +#include +#include +#include +#include +#include "packages/dht/protos/DhtRpc.pb.h" + +import streamr.trackerlessnetwork.ContentDeliveryRpcRemote; +import streamr.trackerlessnetwork.NeighborFinder; +import streamr.trackerlessnetwork.NeighborUpdateRpcLocal; +import streamr.trackerlessnetwork.NetworkRpcClient; +import streamr.trackerlessnetwork.NodeList; +import streamr.trackerlessnetwork.TestUtils; +import streamr.trackerlessnetwork.protos; +import streamr.dht.DhtCallContext; +import streamr.dht.FakeTransport; +import streamr.dht.Identifiers; +import streamr.dht.ListeningRpcCommunicator; +import streamr.dht.protos; +import streamr.utils.StreamPartID; + +using ::dht::PeerDescriptor; +using streamr::dht::DhtAddress; +using streamr::dht::Identifiers; +using streamr::dht::ServiceID; +using streamr::dht::rpcprotocol::DhtCallContext; +using streamr::dht::transport::FakeTransport; +using streamr::dht::transport::ListeningRpcCommunicator; +using streamr::trackerlessnetwork::ContentDeliveryRpcClient; +using streamr::trackerlessnetwork::ContentDeliveryRpcRemote; +using streamr::trackerlessnetwork::NodeList; +using streamr::trackerlessnetwork::neighbordiscovery::INeighborFinder; +using streamr::trackerlessnetwork::neighbordiscovery::NeighborUpdateRpcLocal; +using streamr::trackerlessnetwork::neighbordiscovery:: + NeighborUpdateRpcLocalOptions; +using streamr::trackerlessnetwork::testutils::createMockPeerDescriptor; +using streamr::utils::StreamPartIDUtils; + +namespace { + +constexpr size_t neighborTargetCount = 4; + +// The TS test injects `{ start: jest.fn() }`. +class MockNeighborFinder : public INeighborFinder { +public: + size_t startCalls = 0; + + using INeighborFinder::start; + void start(std::vector /*excluded*/) override { + this->startCalls++; + } + void stop() override {} + [[nodiscard]] bool isRunning() const override { return false; } +}; + +} // namespace + +class NeighborUpdateRpcLocalTest : public ::testing::Test { +protected: + PeerDescriptor localPeerDescriptor = createMockPeerDescriptor(); + FakeTransport transport{ + localPeerDescriptor, [](const auto& /*message*/) {}}; + ListeningRpcCommunicator rpcCommunicator{ServiceID{"mock"}, transport}; + NodeList neighbors{ + Identifiers::getNodeIdFromPeerDescriptor(localPeerDescriptor), + neighborTargetCount + 1}; + NodeList nearbyNodeView{ + Identifiers::getNodeIdFromPeerDescriptor(localPeerDescriptor), + neighborTargetCount}; + MockNeighborFinder neighborFinder; + std::set ongoingHandshakes; + std::optional rpcLocal; + + void SetUp() override { + this->rpcLocal.emplace( + NeighborUpdateRpcLocalOptions{ + .localPeerDescriptor = this->localPeerDescriptor, + .streamPartId = StreamPartIDUtils::parse("stream#0"), + .neighbors = this->neighbors, + .nearbyNodeView = this->nearbyNodeView, + .neighborFinder = this->neighborFinder, + .rpcCommunicator = this->rpcCommunicator, + .neighborTargetCount = neighborTargetCount, + .ongoingHandshakes = this->ongoingHandshakes}); + } + + void addNeighbors(size_t count) { + for (size_t i = 0; i < count; i++) { + ContentDeliveryRpcClient client{this->rpcCommunicator}; + this->neighbors.add( + std::make_shared( + this->localPeerDescriptor, + createMockPeerDescriptor(), + client)); + } + } + + static NeighborUpdate createUpdate( + const std::vector& neighborDescriptors) { + NeighborUpdate update; + update.set_streampartid(StreamPartIDUtils::parse("stream#0")); + for (const auto& descriptor : neighborDescriptors) { + *update.add_neighbordescriptors() = descriptor; + } + update.set_removeme(false); + return update; + } + + static DhtCallContext createContext(const PeerDescriptor& sender) { + DhtCallContext context; + context.incomingSourceDescriptor = sender; + return context; + } +}; + +TEST_F(NeighborUpdateRpcLocalTest, ResponseContainsNeighborListOfExpectedSize) { + this->addNeighbors(neighborTargetCount); + const auto response = this->rpcLocal->neighborUpdate( + createUpdate({this->localPeerDescriptor}), + createContext(createMockPeerDescriptor())); + EXPECT_EQ(response.neighbordescriptors().size(), neighborTargetCount); +} + +TEST_F(NeighborUpdateRpcLocalTest, UpdatesContactsBasedOnCallersNeighbors) { + this->addNeighbors(neighborTargetCount); + EXPECT_EQ(this->nearbyNodeView.size(), 0); + std::vector callerNeighbors; + callerNeighbors.reserve(neighborTargetCount); + for (size_t i = 0; i < neighborTargetCount; i++) { + callerNeighbors.push_back(createMockPeerDescriptor()); + } + this->rpcLocal->neighborUpdate( + createUpdate(callerNeighbors), + createContext(createMockPeerDescriptor())); + EXPECT_EQ(this->nearbyNodeView.size(), 4); +} + +TEST_F(NeighborUpdateRpcLocalTest, DoesNotAskToBeRemovedIfCallerIsNeighbor) { + const auto caller = createMockPeerDescriptor(); + ContentDeliveryRpcClient client{this->rpcCommunicator}; + this->neighbors.add( + std::make_shared( + this->localPeerDescriptor, caller, client)); + const auto response = this->rpcLocal->neighborUpdate( + createUpdate({this->localPeerDescriptor}), createContext(caller)); + EXPECT_EQ(response.removeme(), false); +} + +TEST_F(NeighborUpdateRpcLocalTest, AsksToBeRemovedIfCallerIsNotNeighbor) { + const auto caller = createMockPeerDescriptor(); + const auto response = this->rpcLocal->neighborUpdate( + createUpdate({this->localPeerDescriptor}), createContext(caller)); + EXPECT_EQ(response.removeme(), true); +} + +TEST_F( + NeighborUpdateRpcLocalTest, + AsksToBeRemovedIfCallerIsNeighborAndBothHaveTooManyNeighbors) { + const auto caller = createMockPeerDescriptor(); + ContentDeliveryRpcClient client{this->rpcCommunicator}; + this->neighbors.add( + std::make_shared( + this->localPeerDescriptor, caller, client)); + this->addNeighbors(neighborTargetCount); + std::vector callerNeighbors{this->localPeerDescriptor}; + for (size_t i = 0; i < neighborTargetCount; i++) { + callerNeighbors.push_back(createMockPeerDescriptor()); + } + const auto response = this->rpcLocal->neighborUpdate( + createUpdate(callerNeighbors), createContext(caller)); + EXPECT_EQ(response.removeme(), true); + EXPECT_EQ( + this->neighbors.has(Identifiers::getNodeIdFromPeerDescriptor(caller)), + false); +} + +TEST_F( + NeighborUpdateRpcLocalTest, + DoesNotAskToBeRemovedIfOngoingHandshakeToCaller) { + const auto caller = createMockPeerDescriptor(); + this->ongoingHandshakes.insert( + Identifiers::getNodeIdFromPeerDescriptor(caller)); + const auto response = this->rpcLocal->neighborUpdate( + createUpdate({this->localPeerDescriptor}), createContext(caller)); + EXPECT_EQ(response.removeme(), false); +} diff --git a/packages/streamr-trackerless-network/test/unit/NeighborUpdateRpcRemoteTest.cpp b/packages/streamr-trackerless-network/test/unit/NeighborUpdateRpcRemoteTest.cpp new file mode 100644 index 00000000..5958f6e3 --- /dev/null +++ b/packages/streamr-trackerless-network/test/unit/NeighborUpdateRpcRemoteTest.cpp @@ -0,0 +1,88 @@ +// Ported from packages/trackerless-network/test/integration/ +// NeighborUpdateRpcRemote.test.ts (v103.8.0-rc.3). Adaptation: the +// house two-communicator pattern instead of simulator transports (see +// ContentDeliveryRpcRemoteTest.cpp). +// +// NB: NetworkRpc types are consumed ONLY through the +// streamr.trackerlessnetwork.protos module (no textual NetworkRpc.pb.h +// include) — see TestUtilsTest.cpp for the clangd rationale. +#include +#include +#include +#include "packages/dht/protos/DhtRpc.pb.h" + +#include // IWYU pragma: keep + +import streamr.utils.CoroutineHelper; +import streamr.protorpc.RpcCommunicator; +import streamr.protorpc.protos; +import streamr.trackerlessnetwork.NeighborUpdateRpcRemote; +import streamr.trackerlessnetwork.NetworkRpcClient; +import streamr.trackerlessnetwork.TestUtils; +import streamr.trackerlessnetwork.protos; +import streamr.dht.DhtCallContext; +import streamr.dht.protos; +import streamr.utils.StreamPartID; + +using ::dht::PeerDescriptor; +using ::protorpc::RpcMessage; +using streamr::dht::rpcprotocol::DhtCallContext; +using streamr::protorpc::RpcCommunicator; +using streamr::trackerlessnetwork::neighbordiscovery::NeighborUpdateRpcClient; +using streamr::trackerlessnetwork::neighbordiscovery::NeighborUpdateRpcRemote; +using streamr::trackerlessnetwork::testutils::createMockPeerDescriptor; +using streamr::utils::blockingWait; +using streamr::utils::StreamPartIDUtils; + +using RpcCommunicatorType = RpcCommunicator; + +class NeighborUpdateRpcRemoteTest : public ::testing::Test { +protected: + RpcCommunicatorType clientCommunicator; + RpcCommunicatorType serverCommunicator; + PeerDescriptor clientNode = createMockPeerDescriptor(); + PeerDescriptor serverNode = createMockPeerDescriptor(); + std::optional rpcRemote; + + void SetUp() override { + this->serverCommunicator + .registerRpcMethod( + "neighborUpdate", + [](const NeighborUpdate& /*request*/, + const DhtCallContext& /*context*/) { + NeighborUpdate update; + update.set_streampartid( + StreamPartIDUtils::parse("stream#0")); + *update.add_neighbordescriptors() = + createMockPeerDescriptor(); + update.set_removeme(false); + return update; + }); + this->clientCommunicator.setOutgoingMessageCallback( + [this]( + const RpcMessage& message, + const std::string& /*requestId*/, + const DhtCallContext& /*context*/) { + this->serverCommunicator.handleIncomingMessage( + message, DhtCallContext()); + }); + this->serverCommunicator.setOutgoingMessageCallback( + [this]( + const RpcMessage& message, + const std::string& /*requestId*/, + const DhtCallContext& /*context*/) { + this->clientCommunicator.handleIncomingMessage( + message, DhtCallContext()); + }); + this->rpcRemote.emplace( + this->clientNode, + this->serverNode, + NeighborUpdateRpcClient(this->clientCommunicator)); + } +}; + +TEST_F(NeighborUpdateRpcRemoteTest, UpdateNeighbors) { + const auto result = blockingWait(this->rpcRemote->updateNeighbors( + StreamPartIDUtils::parse("test#0"), {})); + EXPECT_EQ(result.peerDescriptors.size(), 1); +} diff --git a/packages/streamr-trackerless-network/test/utils/TestUtils.cppm b/packages/streamr-trackerless-network/test/utils/TestUtils.cppm index 312d783c..1b104633 100644 --- a/packages/streamr-trackerless-network/test/utils/TestUtils.cppm +++ b/packages/streamr-trackerless-network/test/utils/TestUtils.cppm @@ -9,14 +9,22 @@ module; #include #include +#include #include +#include #include #include "packages/dht/protos/DhtRpc.pb.h" #include "packages/network/protos/NetworkRpc.pb.h" export module streamr.trackerlessnetwork.TestUtils; +import streamr.protorpc.RpcCommunicator; +import streamr.protorpc.protos; +import streamr.trackerlessnetwork.ContentDeliveryRpcRemote; +import streamr.trackerlessnetwork.HandshakeRpcRemote; +import streamr.trackerlessnetwork.NetworkRpcClient; import streamr.dht.ConnectionLocker; +import streamr.dht.DhtCallContext; import streamr.dht.Identifiers; import streamr.utils.BinaryUtils; import streamr.utils.EthereumAddress; @@ -76,6 +84,59 @@ inline PeerDescriptor createMockPeerDescriptor() { return descriptor; } +// Shared transportless communicator behind the mock remote factories +// (the TS factories pass `new RpcCommunicator()` per remote). Outgoing +// messages fail immediately, so any RPC attempted on a mock remote +// reports failure fast instead of waiting for a response timeout. +// NOT leaked, deliberately: the communicator's serial executor holds a +// KeepAlive on the shared worker pool, and a leaked KeepAlive makes the +// pool's static destructor wait forever at process exit +// (DefaultKeepAliveExecutor::joinKeepAlive). As a function-local static +// it is constructed after the pool and therefore destroyed before it, +// releasing the KeepAlive in time. +inline streamr::protorpc::RpcCommunicator< + streamr::dht::rpcprotocol::DhtCallContext>& +getMockRpcCommunicator() { + using streamr::dht::rpcprotocol::DhtCallContext; + static streamr::protorpc::RpcCommunicator communicator; + static const bool initialized = []() { + communicator.setOutgoingMessageCallback( + [](const ::protorpc::RpcMessage& /*message*/, + const std::string& /*requestId*/, + const DhtCallContext& /*context*/) { + throw std::runtime_error( + "mock rpc communicator has no transport"); + }); + return true; + }(); + (void)initialized; + return communicator; +} + +// Ported from createMockContentDeliveryRpcRemote() (test/utils/utils.ts). +inline std::shared_ptr +createMockContentDeliveryRpcRemote( + std::optional remotePeerDescriptor = std::nullopt) { + streamr::trackerlessnetwork::ContentDeliveryRpcClient client{ + getMockRpcCommunicator()}; + return std::make_shared< + streamr::trackerlessnetwork::ContentDeliveryRpcRemote>( + createMockPeerDescriptor(), + remotePeerDescriptor.value_or(createMockPeerDescriptor()), + client); +} + +// Ported from createMockHandshakeRpcRemote() (test/utils/utils.ts). +inline std::shared_ptr< + streamr::trackerlessnetwork::neighbordiscovery::HandshakeRpcRemote> +createMockHandshakeRpcRemote() { + streamr::trackerlessnetwork::neighbordiscovery::HandshakeRpcClient client{ + getMockRpcCommunicator()}; + return std::make_shared< + streamr::trackerlessnetwork::neighbordiscovery::HandshakeRpcRemote>( + createMockPeerDescriptor(), createMockPeerDescriptor(), client); +} + // Ported from mockConnectionLocker (test/utils/utils.ts): a no-op // ConnectionLocker for components that require one but whose locking // behavior is irrelevant to the test. diff --git a/packages/streamr-utils/modules/CoroutineHelper.cppm b/packages/streamr-utils/modules/CoroutineHelper.cppm index a7d8194c..54818fe9 100644 --- a/packages/streamr-utils/modules/CoroutineHelper.cppm +++ b/packages/streamr-utils/modules/CoroutineHelper.cppm @@ -44,6 +44,7 @@ using folly::coro::CancellableAsyncScope; using folly::coro::co_invoke; using folly::coro::collectAll; using folly::coro::collectAllRange; +using folly::coro::collectAllTryRange; using folly::coro::detachOnCancel; using folly::coro::Future; using folly::coro::makePromiseContract;