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;