From 0213c6cad4a442fa3a5d65539209c7b5322f9994 Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Sun, 12 Jul 2026 17:46:05 +0300 Subject: [PATCH] =?UTF-8?q?Phase=20C3:=20ContentDeliveryLayerNode=20?= =?UTF-8?q?=E2=80=94=20the=20per-stream-part=20overlay=20node?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ports the content-delivery overlay node and its factory from the TS pin, composing the C1/C2 classes with propagation, inspection (C7) and the discovery layer (C4): - ContentDeliveryLayerNode: neighbor list + four contact views fed from discovery-layer events, broadcast with per-publisher duplicate detection, leave-notice handling, RPC server wiring, and TS-parity stop() (leave notices + rpcCommunicator destroy). Plumtree branches are milestone-E scope and omitted. - createContentDeliveryLayerNode: defaults taken from the TS source (neighbor target 4, view size 20, min propagation targets 2, 150-message/10 s buffer, 10 s update interval); components are shared_ptr-wired with documented destruction order. - DhtNodeDiscoveryLayer adapter + DhtNode contact-event re-emission: TS passes a DhtNode wherever a DiscoveryLayerNode is expected (structural typing); DhtNode now re-emits the six PeerManager contact events like TS (second event-emitter base, merged lookup) and the adapter provides the interface nominally. Five real defects found and fixed by the ported tests (all verified with before/after runs per the runtime-evidence methodology): 1. NodeList data race: worker threads iterated lists freed by concurrent replaceAll/add (64-node SetUp SIGSEGV, crash-report stack). All access is now mutex-guarded; NodeAdded/NodeRemoved emit OUTSIDE the lock (their handlers make blocking RPC sends). 2. NodeList ordering parity: the TS NodeList is insertion-ordered and the interleave protocol depends on it (getLast = newest neighbor); the std::map-backed port was id-ordered. Now an insertion-ordered vector; the TS ordering test is ported. 3. Teardown lazy-task lifetime: stop()'s leave-notice coroutines captured remotes whose last references concurrent handlers could drop before the batch ran (the rpc-remote lazy-task trap; teardown SIGSEGV). The batch now pins the remotes. 4. previousMessageRef optionality: TS passes undefined when absent; the port forwarded the proto3 default instance, violating the DuplicateMessageDetector invariant for (0,0) refs. Now optional through the chain. 5. Synchronous propagation deadlock: sendToNeighbor ran blockingWait inside the RPC delivery path — under broadcast fan-out nodes blocked sending to each other in cycles and propagation stalled (0/3 timeouts at 64 nodes). Propagation sends are now Task-returning scope tasks (3/3 after); ProxyClient keeps its synchronous result API via an awaitable collect variant. Tests ported: unit/ContentDeliveryLayerNode.test.ts (+ loopback suppression cases), integration/ContentDeliveryLayerNode-Layer1Node .test.ts and its -Latencies variant (shared fixture, 50 ms fixed latency), integration/Propagation.test.ts (64 nodes instead of TS's 256 — the C++ simulator's per-node cost makes 256 a ~16 min run, beyond CI budget; scale-up is a documented follow-up), and the full unit/NodeList.test.ts (14 cases). Local: trackerless-network 83/83 with clean process exit, streamr-dht 181/181, full-tree build green, both package lints rc=0. Co-Authored-By: Claude Fable 5 --- packages/streamr-dht/modules/dht/DhtNode.cppm | 59 +- .../CMakeLists.txt | 3 + packages/streamr-trackerless-network/lint.sh | 5 +- .../DhtNodeDiscoveryLayer.cppm | 110 ++++ .../logic/ContentDeliveryLayerNode.cppm | 600 ++++++++++++++++++ .../logic/ContentDeliveryRpcLocal.cppm | 13 +- .../modules/logic/NodeList.cppm | 137 ++-- .../logic/createContentDeliveryLayerNode.cppm | 265 ++++++++ .../logic/propagation/Propagation.cppm | 162 +++-- .../modules/logic/proxy/ProxyClient.cppm | 42 +- .../ContentDeliveryLayerNodeLayer1Test.cpp | 363 +++++++++++ .../unit/ContentDeliveryLayerNodeTest.cpp | 270 ++++++++ .../test/unit/NodeListTest.cpp | 185 +++++- .../test/unit/PropagationScaleTest.cpp | 226 +++++++ .../test/utils/TestUtils.cppm | 6 + 15 files changed, 2343 insertions(+), 103 deletions(-) create mode 100644 packages/streamr-trackerless-network/modules/discovery-layer/DhtNodeDiscoveryLayer.cppm create mode 100644 packages/streamr-trackerless-network/modules/logic/ContentDeliveryLayerNode.cppm create mode 100644 packages/streamr-trackerless-network/modules/logic/createContentDeliveryLayerNode.cppm create mode 100644 packages/streamr-trackerless-network/test/unit/ContentDeliveryLayerNodeLayer1Test.cpp create mode 100644 packages/streamr-trackerless-network/test/unit/ContentDeliveryLayerNodeTest.cpp create mode 100644 packages/streamr-trackerless-network/test/unit/PropagationScaleTest.cpp diff --git a/packages/streamr-dht/modules/dht/DhtNode.cppm b/packages/streamr-dht/modules/dht/DhtNode.cppm index 2d808686..69924b8c 100644 --- a/packages/streamr-dht/modules/dht/DhtNode.cppm +++ b/packages/streamr-dht/modules/dht/DhtNode.cppm @@ -167,7 +167,34 @@ struct DhtNodeOptions { std::optional maxMessageSize; }; -class DhtNode : public Transport { +// TS DhtNode re-emits the PeerManager contact events as its own events +// (nearbyContactAdded etc.); the trackerless-network discovery-layer +// facade subscribes to them. Mirrored here as a second event-emitter +// base, merged with the Transport events via using-declarations. +using DhtNodeContactEvents = std::tuple< + peermanagerevents::NearbyContactAdded, + peermanagerevents::NearbyContactRemoved, + peermanagerevents::RandomContactAdded, + peermanagerevents::RandomContactRemoved, + peermanagerevents::RingContactAdded, + peermanagerevents::RingContactRemoved>; + +class DhtNode + : public Transport, + public streamr::eventemitter::EventEmitter { +public: + using ContactEventEmitter = + streamr::eventemitter::EventEmitter; + using ContactEventEmitter::emit; + using ContactEventEmitter::off; + using ContactEventEmitter::on; + using ContactEventEmitter::once; + using Transport::emit; + using Transport::off; + using Transport::on; + using Transport::once; + using Transport::removeAllListeners; + private: static constexpr std::chrono::milliseconds externalApiTimeout{10000}; static constexpr std::chrono::milliseconds networkConnectivityPollInterval{ @@ -296,6 +323,36 @@ private: [this](const PeerDescriptor& peerDescriptor) { this->storeManager->onContactAdded(peerDescriptor); }); + // TS re-emits every PeerManager contact event as a DhtNode event. + this->peerManager->on( + [this](const PeerDescriptor& peerDescriptor) { + this->emit( + peerDescriptor); + }); + this->peerManager->on( + [this](const PeerDescriptor& peerDescriptor) { + this->emit( + peerDescriptor); + }); + this->peerManager->on( + [this](const PeerDescriptor& peerDescriptor) { + this->emit( + peerDescriptor); + }); + this->peerManager->on( + [this](const PeerDescriptor& peerDescriptor) { + this->emit( + peerDescriptor); + }); + this->peerManager->on( + [this](const PeerDescriptor& peerDescriptor) { + this->emit(peerDescriptor); + }); + this->peerManager->on( + [this](const PeerDescriptor& peerDescriptor) { + this->emit( + peerDescriptor); + }); this->peerManager->on([this]() { if (!this->peerDiscovery->isJoinOngoing() && !this->options.entryPoints.empty()) { diff --git a/packages/streamr-trackerless-network/CMakeLists.txt b/packages/streamr-trackerless-network/CMakeLists.txt index 216bab89..cc23530f 100644 --- a/packages/streamr-trackerless-network/CMakeLists.txt +++ b/packages/streamr-trackerless-network/CMakeLists.txt @@ -182,6 +182,9 @@ if(NOT IOS AND STREAMR_MODULES_SUPPORTED) test/unit/NeighborFinderTest.cpp test/unit/NeighborUpdateRpcLocalTest.cpp test/unit/NeighborUpdateRpcRemoteTest.cpp + test/unit/ContentDeliveryLayerNodeTest.cpp + test/unit/ContentDeliveryLayerNodeLayer1Test.cpp + test/unit/PropagationScaleTest.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 7c7240ed..a1ddec53 100755 --- a/packages/streamr-trackerless-network/lint.sh +++ b/packages/streamr-trackerless-network/lint.sh @@ -35,8 +35,11 @@ fi # 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. +# ContentDeliveryLayerNodeTest.cpp (phase C3) trips the std-type +# unification false positive on its own std::string locals; 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' | grep -v 'test/unit/HandshakerTest.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' | grep -v 'test/unit/ContentDeliveryLayerNodeTest.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/discovery-layer/DhtNodeDiscoveryLayer.cppm b/packages/streamr-trackerless-network/modules/discovery-layer/DhtNodeDiscoveryLayer.cppm new file mode 100644 index 00000000..49dfeaa5 --- /dev/null +++ b/packages/streamr-trackerless-network/modules/discovery-layer/DhtNodeDiscoveryLayer.cppm @@ -0,0 +1,110 @@ +// Module streamr.trackerlessnetwork.DhtNodeDiscoveryLayer +// The TS code passes a DhtNode wherever a DiscoveryLayerNode is needed +// (structural typing); this adapter provides that shape nominally: it +// implements the DiscoveryLayerNode interface over a layer-1 +// streamr::dht::DhtNode, forwarding the six contact events and +// delegating the getters and lifecycle. +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 "packages/dht/protos/DhtRpc.pb.h" + +export module streamr.trackerlessnetwork.DhtNodeDiscoveryLayer; + +import streamr.utils.CoroutineHelper; +import streamr.trackerlessnetwork.DiscoveryLayerNode; +import streamr.dht.DhtNode; +import streamr.dht.Identifiers; +import streamr.dht.PeerManager; +import streamr.dht.protos; + +using streamr::dht::ClosestRingPeerDescriptors; +using streamr::dht::DhtAddress; +using streamr::dht::DhtNode; + +export namespace streamr::trackerlessnetwork::discoverylayer { + +using ::dht::PeerDescriptor; + +class DhtNodeDiscoveryLayer : public DiscoveryLayerNode { +private: + std::shared_ptr dhtNode; + + template + void forward() { + this->dhtNode->template on( + [this](const PeerDescriptor& peerDescriptor) { + this->template emit(peerDescriptor); + }); + } + +public: + explicit DhtNodeDiscoveryLayer(std::shared_ptr dhtNode) + : dhtNode(std::move(dhtNode)) { + namespace pme = streamr::dht::peermanagerevents; + namespace dle = discoverylayernodeevents; + this->forward(); + this->forward(); + this->forward(); + this->forward(); + this->forward(); + this->forward(); + } + + [[nodiscard]] DhtNode& getDhtNode() { return *this->dhtNode; } + + void removeContact(const DhtAddress& nodeId) override { + this->dhtNode->removeContact(nodeId); + } + + [[nodiscard]] std::vector getClosestContacts( + std::optional maxCount = std::nullopt) override { + return this->dhtNode->getClosestContacts(maxCount); + } + + [[nodiscard]] std::vector getRandomContacts( + std::optional maxCount = std::nullopt) override { + return this->dhtNode->getRandomContacts(maxCount); + } + + [[nodiscard]] ClosestRingPeerDescriptors getRingContacts() override { + return this->dhtNode->getRingContacts(); + } + + [[nodiscard]] std::vector getNeighbors() override { + return this->dhtNode->getNeighbors(); + } + + [[nodiscard]] size_t getNeighborCount() override { + return this->dhtNode->getNeighborCount(); + } + + folly::coro::Task joinDht( + std::vector entryPoints, + bool doRandomJoin = true, + bool retry = true) override { + co_await this->dhtNode->joinDht( + std::move(entryPoints), doRandomJoin, retry); + } + + folly::coro::Task joinRing() override { + co_await this->dhtNode->joinRing(); + } + + folly::coro::Task start() override { + co_await this->dhtNode->start(); + } + + folly::coro::Task stop() override { + this->dhtNode->stop(); + co_return; + } +}; + +} // namespace streamr::trackerlessnetwork::discoverylayer diff --git a/packages/streamr-trackerless-network/modules/logic/ContentDeliveryLayerNode.cppm b/packages/streamr-trackerless-network/modules/logic/ContentDeliveryLayerNode.cppm new file mode 100644 index 00000000..aa49150d --- /dev/null +++ b/packages/streamr-trackerless-network/modules/logic/ContentDeliveryLayerNode.cppm @@ -0,0 +1,600 @@ +// Module streamr.trackerlessnetwork.ContentDeliveryLayerNode +// Ported from packages/trackerless-network/src/content-delivery-layer/ +// ContentDeliveryLayerNode.ts (v103.8.0-rc.3): the per-stream-part +// overlay node — owns the neighbor list and the four contact views fed +// from the discovery layer, broadcasts with per-publisher duplicate +// detection, and wires the content-delivery / temporary-connection RPC +// servers. The plumtree branch is milestone-E scope (plan decision 3.3) +// and the TS GapDiagnostics sampling is a TS-only diagnostic; both are +// omitted. +// +// Adaptations: components arrive as shared_ptrs (the TS factory relies +// on GC; here createContentDeliveryLayerNode builds the ownership graph +// and the options struct keeps referenced parts alive — view lists and +// the ongoing-handshake set are declared before the components that +// hold references into them, so they are destroyed last). The TS +// fire-and-forget leave notices in stop() are awaited as notifications +// (they complete at send-enqueue time and swallow errors). +module; + +#include // IWYU pragma: keep + +#include +#include +#include +#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.ContentDeliveryLayerNode; + +import streamr.utils.CoroutineHelper; +import streamr.trackerlessnetwork.ContentDeliveryRpcLocal; +import streamr.trackerlessnetwork.ContentDeliveryRpcRemote; +import streamr.trackerlessnetwork.DiscoveryLayerNode; +import streamr.trackerlessnetwork.DuplicateMessageDetector; +import streamr.trackerlessnetwork.Handshaker; +import streamr.trackerlessnetwork.Inspector; +import streamr.trackerlessnetwork.NeighborFinder; +import streamr.trackerlessnetwork.NeighborUpdateManager; +import streamr.trackerlessnetwork.NetworkRpcClient; +import streamr.trackerlessnetwork.NodeList; +import streamr.trackerlessnetwork.Propagation; +import streamr.trackerlessnetwork.ProxyConnectionRpcLocal; +import streamr.trackerlessnetwork.TemporaryConnectionRpcLocal; +import streamr.trackerlessnetwork.Utils; +import streamr.dht.ConnectionLocker; +import streamr.dht.DhtCallContext; +import streamr.dht.Identifiers; +import streamr.dht.ListeningRpcCommunicator; +import streamr.dht.Transport; +import streamr.dht.protos; +import streamr.eventemitter.EventEmitter; +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::connection::ConnectionLocker; +using streamr::dht::connection::LockID; +using streamr::dht::rpcprotocol::DhtCallContext; +using streamr::dht::transport::ListeningRpcCommunicator; +using streamr::dht::transport::Transport; +using streamr::eventemitter::Event; +using streamr::eventemitter::EventEmitter; +using streamr::eventemitter::HandlerToken; +using streamr::logger::SLogger; +using streamr::utils::StreamPartID; + +export namespace streamr::trackerlessnetwork { + +using ::dht::PeerDescriptor; +using streamr::trackerlessnetwork::discoverylayer::DiscoveryLayerNode; +using streamr::trackerlessnetwork::inspection::Inspector; +using streamr::trackerlessnetwork::neighbordiscovery::Handshaker; +using streamr::trackerlessnetwork::neighbordiscovery::INeighborFinder; +using streamr::trackerlessnetwork::neighbordiscovery::NeighborUpdateManager; +using streamr::trackerlessnetwork::propagation::Propagation; +using streamr::trackerlessnetwork::proxy::ProxyConnectionRpcLocal; + +namespace contentdeliverylayernodeevents { + +struct Message : Event {}; +struct NeighborConnected : Event {}; +struct EntryPointLeaveDetected : Event<> {}; + +} // namespace contentdeliverylayernodeevents + +using ContentDeliveryLayerNodeEvents = std::tuple< + contentdeliverylayernodeevents::Message, + contentdeliverylayernodeevents::NeighborConnected, + contentdeliverylayernodeevents::EntryPointLeaveDetected>; + +constexpr size_t defaultNodeViewSize = 20; +constexpr size_t defaultNeighborTargetCount = 4; +constexpr bool defaultAcceptProxyConnections = false; + +struct ContentDeliveryLayerNeighborInfo { + PeerDescriptor peerDescriptor; + std::optional rtt; +}; + +// Members referenced by other components (view lists, the +// ongoing-handshake set, the communicator) are declared FIRST so they +// outlive the components holding references into them. +struct StrictContentDeliveryLayerNodeOptions { + StreamPartID streamPartId; + std::shared_ptr discoveryLayerNode; + Transport* transport = nullptr; + ConnectionLocker* connectionLocker = nullptr; + PeerDescriptor localPeerDescriptor; + size_t nodeViewSize = defaultNodeViewSize; + std::shared_ptr rpcCommunicator; + std::shared_ptr> ongoingHandshakes; + std::shared_ptr nearbyNodeView; + std::shared_ptr randomNodeView; + std::shared_ptr leftNodeView; + std::shared_ptr rightNodeView; + std::shared_ptr neighbors; + std::shared_ptr temporaryConnectionRpcLocal; + std::shared_ptr + proxyConnectionRpcLocal; // may be null + std::shared_ptr propagation; + std::shared_ptr handshaker; + std::shared_ptr neighborFinder; + std::shared_ptr neighborUpdateManager; + std::shared_ptr inspector; + size_t neighborTargetCount = defaultNeighborTargetCount; + std::function isLocalNodeEntryPoint; + std::optional rpcRequestTimeout; + bool suppressOwnMessageLoopback = false; +}; + +class ContentDeliveryLayerNode + : public EventEmitter { +private: + StrictContentDeliveryLayerNodeOptions options; + std::map duplicateDetectors; + std::optional contentDeliveryRpcLocal; + std::atomic started = false; + std::atomic stopped = false; + std::vector> unsubscribers; + std::atomic messagesPropagated = 0; + +public: + explicit ContentDeliveryLayerNode( + StrictContentDeliveryLayerNodeOptions options) + : options(std::move(options)) { + this->contentDeliveryRpcLocal.emplace( + ContentDeliveryRpcLocalOptions{ + .localPeerDescriptor = this->options.localPeerDescriptor, + .streamPartId = this->options.streamPartId, + .markAndCheckDuplicate = + [this]( + const MessageID& msg, + const std::optional& prev) { + return Utils::markAndCheckDuplicate( + this->duplicateDetectors, msg, prev); + }, + .broadcast = + [this]( + const StreamMessage& message, + const DhtAddress& previousNode) { + this->broadcast(message, previousNode); + }, + .onLeaveNotice = + [this]( + const DhtAddress& remoteNodeId, + bool sourceIsStreamEntryPoint) { + this->onLeaveNotice( + remoteNodeId, sourceIsStreamEntryPoint); + }, + .markForInspection = + [this]( + const DhtAddress& remoteNodeId, + const MessageID& messageId) { + this->options.inspector->markMessage( + remoteNodeId, messageId); + }, + .rpcCommunicator = *this->options.rpcCommunicator}); + } + + ~ContentDeliveryLayerNode() override { this->stop(); } + + folly::coro::Task start() { + this->started = true; + this->registerDefaultServerMethods(); + this->subscribeToSources(); + this->options.neighborFinder->start(); + this->options.neighborUpdateManager->start(); + co_return; + } + + void stop() { + if (!this->started || this->stopped.exchange(true)) { + return; + } + for (const auto& unsubscribe : this->unsubscribers) { + unsubscribe(); + } + this->unsubscribers.clear(); + if (this->options.proxyConnectionRpcLocal) { + this->options.proxyConnectionRpcLocal->stop(); + } + { + // TS fires these without awaiting; the notices are + // notifications (complete at send-enqueue) and swallow + // errors, so awaiting them here is bounded. + // Pin the remotes for the whole batch: the lazy notice + // coroutines capture each remote's `this`, and concurrent + // handlers may drop the neighbor list's own references + // before blockingWait starts the tasks (the 64-node + // teardown SIGSEGV — the rpc-remote lazy-task trap). + const auto remotes = this->options.neighbors->getAll(); + std::vector> notices; + notices.reserve(remotes.size()); + for (const auto& remote : remotes) { + notices.push_back(remote->leaveStreamPartNotice( + this->options.streamPartId, + this->options.isLocalNodeEntryPoint())); + this->options.connectionLocker->weakUnlockConnection( + Identifiers::getNodeIdFromPeerDescriptor( + remote->getPeerDescriptor()), + LockID{this->options.streamPartId}); + } + streamr::utils::blockingWait( + folly::coro::collectAllTryRange(std::move(notices))); + } + // TS parity: destroy() unregisters from the transport and drains + // the communicator scopes NOW, while the send targets are alive — + // a straggler leave-notice coroutine resuming after this object + // dies was the 64-node teardown SIGSEGV. + this->options.rpcCommunicator->destroy(); + this->removeAllListeners(); + this->options.nearbyNodeView->stop(); + this->options.neighbors->stop(); + this->options.randomNodeView->stop(); + this->options.neighborFinder->stop(); + this->options.neighborUpdateManager->stop(); + this->options.inspector->stop(); + this->duplicateDetectors.clear(); + } + + void broadcast( + const StreamMessage& msg, + const std::optional& previousNode = std::nullopt) { + if (!previousNode.has_value()) { + Utils::markAndCheckDuplicate( + this->duplicateDetectors, + msg.messageid(), + msg.has_previousmessageref() + ? std::optional(msg.previousmessageref()) + : std::nullopt); + } + // Deliver to local listeners — except own publishes (no + // previousNode) when loopback is suppressed: nothing local + // consumes them. Propagation + duplicate detection are + // unaffected. + if (previousNode.has_value() || + !this->options.suppressOwnMessageLoopback) { + this->emit(msg); + } + const bool skipBackPropagation = previousNode.has_value() && + !this->options.temporaryConnectionRpcLocal->hasNode( + previousNode.value()); + this->options.propagation->feedUnseenMessage( + msg, + this->getPropagationTargets(msg), + skipBackPropagation ? previousNode : std::nullopt); + this->messagesPropagated++; + } + + folly::coro::Task inspect(PeerDescriptor peerDescriptor) { + return this->options.inspector->inspect(std::move(peerDescriptor)); + } + + [[nodiscard]] bool hasProxyConnection(const DhtAddress& nodeId) const { + if (this->options.proxyConnectionRpcLocal) { + return this->options.proxyConnectionRpcLocal->hasConnection(nodeId); + } + return false; + } + + [[nodiscard]] DhtAddress getOwnNodeId() const { + return Identifiers::getNodeIdFromPeerDescriptor( + this->options.localPeerDescriptor); + } + + [[nodiscard]] size_t getOutgoingHandshakeCount() const { + return this->options.handshaker->getOngoingHandshakes().size(); + } + + [[nodiscard]] std::vector getNeighbors() const { + if (!this->started && this->stopped) { + return {}; + } + std::vector descriptors; + for (const auto& neighbor : this->options.neighbors->getAll()) { + descriptors.push_back(neighbor->getPeerDescriptor()); + } + return descriptors; + } + + [[nodiscard]] std::vector getInfos() + const { + std::vector infos; + for (const auto& neighbor : this->options.neighbors->getAll()) { + infos.push_back( + ContentDeliveryLayerNeighborInfo{ + .peerDescriptor = neighbor->getPeerDescriptor(), + .rtt = neighbor->getRtt()}); + } + return infos; + } + + [[nodiscard]] NodeList& getNearbyNodeView() { + return *this->options.nearbyNodeView; + } + +private: + void registerDefaultServerMethods() { + this->options.rpcCommunicator->registerRpcNotification( + "sendStreamMessage", + [this](const StreamMessage& msg, const DhtCallContext& context) { + this->contentDeliveryRpcLocal->sendStreamMessage(msg, context); + }); + this->options.rpcCommunicator + ->registerRpcNotification( + "leaveStreamPartNotice", + [this]( + const LeaveStreamPartNotice& req, + const DhtCallContext& context) { + this->contentDeliveryRpcLocal->leaveStreamPartNotice( + req, context); + }); + this->options.rpcCommunicator->registerRpcMethod< + TemporaryConnectionRequest, + TemporaryConnectionResponse>( + "openConnection", + [this]( + const TemporaryConnectionRequest& req, + const DhtCallContext& context) { + return this->options.temporaryConnectionRpcLocal + ->openConnection(req, context); + }); + this->options.rpcCommunicator + ->registerRpcNotification( + "closeConnection", + [this]( + const CloseTemporaryConnection& req, + const DhtCallContext& context) { + this->options.temporaryConnectionRpcLocal->closeConnection( + req, context); + }); + } + + template + void subscribe(EmitterType& source, Handler handler) { + const auto token = source.template on(std::move(handler)); + this->unsubscribers.emplace_back( + [&source, token]() { source.template off(token); }); + } + + void subscribeToSources() { + namespace dle = streamr::trackerlessnetwork::discoverylayer:: + discoverylayernodeevents; + auto& discovery = *this->options.discoveryLayerNode; + this->subscribe( + discovery, [this](const PeerDescriptor& /*peer*/) { + this->onNearbyContactAdded(); + }); + this->subscribe( + discovery, [this](const PeerDescriptor& /*peer*/) { + this->onNearbyContactRemoved(); + }); + this->subscribe( + discovery, [this](const PeerDescriptor& /*peer*/) { + this->onRandomContactAdded(); + }); + this->subscribe( + discovery, [this](const PeerDescriptor& /*peer*/) { + this->onRandomContactRemoved(); + }); + this->subscribe( + discovery, [this](const PeerDescriptor& /*peer*/) { + this->onRingContactsUpdated(); + }); + this->subscribe( + discovery, [this](const PeerDescriptor& /*peer*/) { + this->onRingContactsUpdated(); + }); + this->subscribe( + *this->options.transport, + [this]( + const PeerDescriptor& peerDescriptor, bool /*gracefulLeave*/) { + this->onNodeDisconnected(peerDescriptor); + }); + this->subscribe( + *this->options.neighbors, + [this]( + const DhtAddress& id, + const std::shared_ptr& remote) { + this->options.propagation->onNeighborJoined(id); + this->options.connectionLocker->weakLockConnection( + Identifiers::getNodeIdFromPeerDescriptor( + remote->getPeerDescriptor()), + LockID{this->options.streamPartId}); + this->emit( + id); + }); + this->subscribe( + *this->options.neighbors, + [this]( + const DhtAddress& /*id*/, + const std::shared_ptr& remote) { + this->options.connectionLocker->weakUnlockConnection( + Identifiers::getNodeIdFromPeerDescriptor( + remote->getPeerDescriptor()), + LockID{this->options.streamPartId}); + }); + if (this->options.proxyConnectionRpcLocal) { + this->subscribe( + *this->options.proxyConnectionRpcLocal, + [this](const DhtAddress& id) { + this->options.propagation->onNeighborJoined(id); + }); + } + } + + void onLeaveNotice( + const DhtAddress& remoteNodeId, bool sourceIsStreamEntryPoint) { + if (this->stopped) { + return; + } + const bool isKnown = + this->options.nearbyNodeView->get(remoteNodeId).has_value() || + this->options.randomNodeView->get(remoteNodeId).has_value() || + this->options.neighbors->get(remoteNodeId).has_value() || + (this->options.proxyConnectionRpcLocal != nullptr && + this->options.proxyConnectionRpcLocal->getConnection(remoteNodeId) + .has_value()); + // TS TODO preserved: check integrity of notifier? + if (isKnown) { + this->options.discoveryLayerNode->removeContact(remoteNodeId); + this->options.neighbors->remove(remoteNodeId); + this->options.nearbyNodeView->remove(remoteNodeId); + this->options.randomNodeView->remove(remoteNodeId); + this->options.leftNodeView->remove(remoteNodeId); + this->options.rightNodeView->remove(remoteNodeId); + this->options.neighborFinder->start({remoteNodeId}); + if (this->options.proxyConnectionRpcLocal) { + this->options.proxyConnectionRpcLocal->removeConnection( + remoteNodeId); + } + } + if (sourceIsStreamEntryPoint) { + this->emit< + contentdeliverylayernodeevents::EntryPointLeaveDetected>(); + } + } + + std::shared_ptr createRemote( + const PeerDescriptor& peer) { + ContentDeliveryRpcClient client{*this->options.rpcCommunicator}; + return std::make_shared( + this->options.localPeerDescriptor, + peer, + client, + this->options.rpcRequestTimeout); + } + + void onRingContactsUpdated() { + SLogger::trace("onRingContactsUpdated"); + if (this->stopped) { + return; + } + const auto contacts = + this->options.discoveryLayerNode->getRingContacts(); + std::vector> left; + for (const auto& peer : contacts.left) { + left.push_back(this->createRemote(peer)); + } + this->options.leftNodeView->replaceAll(left); + std::vector> right; + for (const auto& peer : contacts.right) { + right.push_back(this->createRemote(peer)); + } + this->options.rightNodeView->replaceAll(right); + } + + void onNearbyContactAdded() { + SLogger::trace("New nearby contact found"); + if (this->stopped) { + return; + } + this->updateNearbyNodeView( + this->options.discoveryLayerNode->getClosestContacts( + this->options.nodeViewSize)); + if (this->options.neighbors->size() < + this->options.neighborTargetCount) { + this->options.neighborFinder->start(); + } + } + + void onNearbyContactRemoved() { + SLogger::trace("Nearby contact removed"); + if (this->stopped) { + return; + } + this->updateNearbyNodeView( + this->options.discoveryLayerNode->getClosestContacts( + this->options.nodeViewSize)); + } + + void updateNearbyNodeView(const std::vector& nodes) { + std::vector> remotes; + for (const auto& descriptor : nodes) { + remotes.push_back(this->createRemote(descriptor)); + } + this->options.nearbyNodeView->replaceAll(remotes); + for (const auto& descriptor : + this->options.discoveryLayerNode->getNeighbors()) { + if (this->options.nearbyNodeView->size() >= + this->options.nodeViewSize) { + break; + } + this->options.nearbyNodeView->add(this->createRemote(descriptor)); + } + } + + void onRandomContactAdded() { + if (this->stopped) { + return; + } + this->replaceRandomNodeView(); + if (this->options.neighbors->size() < + this->options.neighborTargetCount) { + this->options.neighborFinder->start(); + } + } + + void onRandomContactRemoved() { + SLogger::trace("New random contact removed"); + if (this->stopped) { + return; + } + this->replaceRandomNodeView(); + } + + void replaceRandomNodeView() { + const auto randomContacts = + this->options.discoveryLayerNode->getRandomContacts( + this->options.nodeViewSize); + std::vector> remotes; + for (const auto& descriptor : randomContacts) { + remotes.push_back(this->createRemote(descriptor)); + } + this->options.randomNodeView->replaceAll(remotes); + } + + void onNodeDisconnected(const PeerDescriptor& peerDescriptor) { + const auto nodeId = + Identifiers::getNodeIdFromPeerDescriptor(peerDescriptor); + if (this->options.neighbors->has(nodeId)) { + this->options.neighbors->remove(nodeId); + this->options.neighborFinder->start({nodeId}); + this->options.temporaryConnectionRpcLocal->removeNode(nodeId); + } + } + + std::vector getPropagationTargets(const StreamMessage& msg) { + auto propagationTargets = this->options.neighbors->getIds(); + if (this->options.proxyConnectionRpcLocal) { + const auto proxyTargets = + this->options.proxyConnectionRpcLocal->getPropagationTargets( + msg); + propagationTargets.insert( + propagationTargets.end(), + proxyTargets.begin(), + proxyTargets.end()); + } + const auto temporaryTargets = + this->options.temporaryConnectionRpcLocal->getNodes().getIds(); + propagationTargets.insert( + propagationTargets.end(), + temporaryTargets.begin(), + temporaryTargets.end()); + return propagationTargets; + } +}; + +} // namespace streamr::trackerlessnetwork diff --git a/packages/streamr-trackerless-network/modules/logic/ContentDeliveryRpcLocal.cppm b/packages/streamr-trackerless-network/modules/logic/ContentDeliveryRpcLocal.cppm index 0034a171..ae6e7335 100644 --- a/packages/streamr-trackerless-network/modules/logic/ContentDeliveryRpcLocal.cppm +++ b/packages/streamr-trackerless-network/modules/logic/ContentDeliveryRpcLocal.cppm @@ -3,6 +3,8 @@ // (MODERNIZATION.md Phase 2.6): this file is now the source of truth. module; +#include + #include "packages/dht/protos/DhtRpc.pb.h" #include "packages/network/protos/NetworkRpc.pb.h" @@ -32,7 +34,11 @@ using ContentDeliveryRpc = struct ContentDeliveryRpcLocalOptions { PeerDescriptor localPeerDescriptor; StreamPartID streamPartId; - std::function + // TS passes `undefined` when the message has no previousMessageRef; + // the proto3 default instance (0, 0) must not be forwarded (it + // violates the DuplicateMessageDetector invariant when the current + // ref is also (0, 0)). + std::function&)> markAndCheckDuplicate; std::function broadcast; std::function onLeaveNotice; @@ -54,7 +60,10 @@ public: context.incomingSourceDescriptor.value()); this->options.markForInspection(previousNode, message.messageid()); if (this->options.markAndCheckDuplicate( - message.messageid(), message.previousmessageref())) { + message.messageid(), + message.has_previousmessageref() + ? std::optional(message.previousmessageref()) + : std::nullopt)) { this->options.broadcast(message, previousNode); } } diff --git a/packages/streamr-trackerless-network/modules/logic/NodeList.cppm b/packages/streamr-trackerless-network/modules/logic/NodeList.cppm index 34739651..06c3dba7 100644 --- a/packages/streamr-trackerless-network/modules/logic/NodeList.cppm +++ b/packages/streamr-trackerless-network/modules/logic/NodeList.cppm @@ -1,9 +1,27 @@ // Module streamr.trackerlessnetwork.NodeList // CONSOLIDATED from the former header logic/NodeList.hpp // (MODERNIZATION.md Phase 2.6): this file is now the source of truth. +// +// Storage is INSERTION-ORDERED (phase C3): the TS NodeList is a Map, +// whose iteration order is insertion order, and the protocol depends on +// it — getLast() must return the most recently added node (the handshake +// interleave hands the newest neighbor over), not the id-largest one as +// the earlier std::map-backed port did. Lookups are linear; the lists +// are bounded by small limits (default 20). +// +// Thread safety (phase C3): TS runs single-threaded; here RPC handler +// threads, the neighbor-finder chains, the neighbor-update interval and +// discovery-event handlers all mutate the same lists concurrently (a +// 64-node scale test crashed reading entries freed by a concurrent +// replaceAll). All container access is mutex-guarded; NodeAdded / +// NodeRemoved are emitted OUTSIDE the lock because their handlers make +// blocking RPC sends. module; -#include +#include +#include +#include +#include #include #include @@ -33,18 +51,32 @@ struct NodeRemoved using NodeListEvents = std::tuple; -// The items in the list are in the insertion order - class NodeList : public EventEmitter { private: - std::map> nodes; + using Entry = + std::pair>; + + mutable std::mutex mutex; + std::vector nodes; size_t limit; DhtAddress ownId; + [[nodiscard]] std::vector::const_iterator find( + const DhtAddress& nodeId) const { + return std::ranges::find_if(this->nodes, [&nodeId](const auto& entry) { + return entry.first == nodeId; + }); + } + + [[nodiscard]] std::vector::iterator find(const DhtAddress& nodeId) { + return std::ranges::find_if(this->nodes, [&nodeId](const auto& entry) { + return entry.first == nodeId; + }); + } + static std::vector> getValuesOfIncludedKeys( - const std::map>& - nodes, + const std::vector& nodes, const std::vector& exclude, bool wsOnly = false) { return nodes | std::views::values | @@ -69,48 +101,77 @@ public: void add(const std::shared_ptr& remote) { const auto nodeId = Identifiers::getNodeIdFromPeerDescriptor( remote->getPeerDescriptor()); - if ((this->ownId != nodeId) && (this->nodes.size() < this->limit)) { - const auto isExistingNode = this->nodes.contains(nodeId); - this->nodes[nodeId] = remote; - - if (!isExistingNode) { - this->emit(nodeId, remote); + bool added = false; + { + std::scoped_lock lock(this->mutex); + if ((this->ownId != nodeId) && (this->nodes.size() < this->limit)) { + const auto it = this->find(nodeId); + if (it != this->nodes.end()) { + it->second = remote; + } else { + this->nodes.emplace_back(nodeId, remote); + added = true; + } } } + if (added) { + this->emit(nodeId, remote); + } } void remove(const DhtAddress& nodeId) { - if (this->nodes.contains(nodeId)) { - const auto remote = this->nodes.at(nodeId); - this->nodes.erase(nodeId); - this->emit(nodeId, remote); + std::shared_ptr removed; + { + std::scoped_lock lock(this->mutex); + const auto it = this->find(nodeId); + if (it != this->nodes.end()) { + removed = it->second; + this->nodes.erase(it); + } + } + if (removed) { + this->emit(nodeId, removed); } } [[nodiscard]] bool has(const DhtAddress& nodeId) const { - return this->nodes.contains(nodeId); + std::scoped_lock lock(this->mutex); + return this->find(nodeId) != this->nodes.end(); } // Replace nodes does not emit nodeRemoved events, use with caution void replaceAll( const std::vector>& neighbors) { - this->nodes.clear(); - const auto limited = - neighbors | std::views::take(this->limit) | - std::ranges::to< - std::vector>>(); - for (const auto& remote : limited) { - this->add(remote); + std::vector addedEntries; + { + std::scoped_lock lock(this->mutex); + this->nodes.clear(); + for (const auto& remote : + neighbors | std::views::take(this->limit)) { + const auto nodeId = Identifiers::getNodeIdFromPeerDescriptor( + remote->getPeerDescriptor()); + if ((this->ownId != nodeId) && + (this->nodes.size() < this->limit) && + this->find(nodeId) == this->nodes.end()) { + this->nodes.emplace_back(nodeId, remote); + addedEntries.emplace_back(nodeId, remote); + } + } + } + for (const auto& [nodeId, remote] : addedEntries) { + this->emit(nodeId, remote); } } [[nodiscard]] std::vector getIds() const { + std::scoped_lock lock(this->mutex); return this->nodes | std::views::keys | std::ranges::to>(); } [[nodiscard]] std::vector getPeerDescriptors() const { + std::scoped_lock lock(this->mutex); return this->nodes | std::views::values | std::views::transform([](const auto& remote) { return remote->getPeerDescriptor(); @@ -118,11 +179,11 @@ public: std::ranges::to>(); } - // TS getNode() returns undefined for unknown ids; map::at would - // throw instead, so look up with find. + // TS getNode() returns undefined for unknown ids. [[nodiscard]] std::optional> get( const DhtAddress& id) const { - if (const auto it = this->nodes.find(id); it != this->nodes.end()) { + std::scoped_lock lock(this->mutex); + if (const auto it = this->find(id); it != this->nodes.end()) { return it->second; } return std::nullopt; @@ -130,11 +191,13 @@ public: [[nodiscard]] size_t size( const std::vector& exclude = {}) const { + std::scoped_lock lock(this->mutex); return NodeList::getValuesOfIncludedKeys(this->nodes, exclude).size(); } [[nodiscard]] std::optional> getRandom(const std::vector& exclude) { + std::scoped_lock lock(this->mutex); const auto values = NodeList::getValuesOfIncludedKeys(this->nodes, exclude); if (values.empty()) { @@ -145,6 +208,7 @@ public: [[nodiscard]] std::optional> getFirst(const std::vector& exclude, bool wsOnly = false) { + std::scoped_lock lock(this->mutex); const auto included = NodeList::getValuesOfIncludedKeys(this->nodes, exclude, wsOnly); return included.empty() ? std::nullopt @@ -153,6 +217,7 @@ public: [[nodiscard]] std::vector> getFirstAndLast(const std::vector& exclude) { + std::scoped_lock lock(this->mutex); const auto included = NodeList::getValuesOfIncludedKeys(this->nodes, exclude); if (included.empty()) { @@ -164,25 +229,19 @@ public: return {included.front()}; } - [[nodiscard]] static std::optional< - std::shared_ptr> - getLast( - const std::map>& - nodes, - const std::vector& exclude) { - const auto included = NodeList::getValuesOfIncludedKeys(nodes, exclude); - return included.empty() ? std::nullopt - : 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); + std::scoped_lock lock(this->mutex); + const auto included = + NodeList::getValuesOfIncludedKeys(this->nodes, exclude); + return included.empty() ? std::nullopt + : std::make_optional(included.back()); } [[nodiscard]] std::vector> getAll() { + std::scoped_lock lock(this->mutex); return this->nodes | std::views::values | std::ranges::to< std::vector>>(); diff --git a/packages/streamr-trackerless-network/modules/logic/createContentDeliveryLayerNode.cppm b/packages/streamr-trackerless-network/modules/logic/createContentDeliveryLayerNode.cppm new file mode 100644 index 00000000..e51613ec --- /dev/null +++ b/packages/streamr-trackerless-network/modules/logic/createContentDeliveryLayerNode.cppm @@ -0,0 +1,265 @@ +// Module streamr.trackerlessnetwork.createContentDeliveryLayerNode +// Ported from packages/trackerless-network/src/content-delivery-layer/ +// createContentDeliveryLayerNode.ts (v103.8.0-rc.3): fills the optional +// components of ContentDeliveryLayerNode with defaults built from the +// exact TS default values (neighbor target 4, view size 20, min +// propagation targets 2, buffer 150 / 10 s, update interval 10 s). +// The plumtree branch is milestone-E scope and omitted; the TS +// bufferWhileConnecting send flag is not plumbed through the C++ +// send options yet (documented deviation — propagation retries cover +// the connecting window). +module; + +#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.createContentDeliveryLayerNode; + +import streamr.utils.CoroutineHelper; +import streamr.trackerlessnetwork.ContentDeliveryLayerNode; +import streamr.trackerlessnetwork.DiscoveryLayerNode; +import streamr.trackerlessnetwork.formStreamPartDeliveryServiceId; +import streamr.trackerlessnetwork.Handshaker; +import streamr.trackerlessnetwork.Inspector; +import streamr.trackerlessnetwork.NeighborFinder; +import streamr.trackerlessnetwork.NeighborUpdateManager; +import streamr.trackerlessnetwork.NodeList; +import streamr.trackerlessnetwork.Propagation; +import streamr.trackerlessnetwork.ProxyConnectionRpcLocal; +import streamr.trackerlessnetwork.TemporaryConnectionRpcLocal; +import streamr.dht.ConnectionLocker; +import streamr.dht.Identifiers; +import streamr.dht.ListeningRpcCommunicator; +import streamr.dht.Transport; +import streamr.dht.protos; +import streamr.utils.StreamPartID; + +using streamr::dht::DhtAddress; +using streamr::dht::Identifiers; +using streamr::dht::connection::ConnectionLocker; +using streamr::dht::transport::ListeningRpcCommunicator; +using streamr::dht::transport::Transport; +using streamr::utils::StreamPartID; + +export namespace streamr::trackerlessnetwork { + +using ::dht::PeerDescriptor; +using streamr::trackerlessnetwork::discoverylayer::DiscoveryLayerNode; +using streamr::trackerlessnetwork::inspection::Inspector; +using streamr::trackerlessnetwork::inspection::InspectorOptions; +using streamr::trackerlessnetwork::neighbordiscovery:: + defaultNeighborUpdateInterval; +using streamr::trackerlessnetwork::neighbordiscovery::Handshaker; +using streamr::trackerlessnetwork::neighbordiscovery::HandshakerOptions; +using streamr::trackerlessnetwork::neighbordiscovery::INeighborFinder; +using streamr::trackerlessnetwork::neighbordiscovery::NeighborFinder; +using streamr::trackerlessnetwork::neighbordiscovery::NeighborFinderOptions; +using streamr::trackerlessnetwork::neighbordiscovery::NeighborUpdateManager; +using streamr::trackerlessnetwork::neighbordiscovery:: + NeighborUpdateManagerOptions; +using streamr::trackerlessnetwork::propagation::DEFAULT_MAX_MESSAGES; +using streamr::trackerlessnetwork::propagation::DEFAULT_TTL; +using streamr::trackerlessnetwork::propagation::Propagation; +using streamr::trackerlessnetwork::propagation::PropagationOptions; +using streamr::trackerlessnetwork::proxy::ProxyConnectionRpcLocal; +using streamr::trackerlessnetwork::proxy::ProxyConnectionRpcLocalOptions; + +constexpr size_t defaultMinPropagationTargets = 2; + +struct ContentDeliveryLayerNodeOptions { + StreamPartID streamPartId; + std::shared_ptr discoveryLayerNode; + Transport* transport = nullptr; + ConnectionLocker* connectionLocker = nullptr; + PeerDescriptor localPeerDescriptor; + std::function isLocalNodeEntryPoint; + // Optional component overrides (tests inject doubles here). + std::shared_ptr rpcCommunicator; + std::shared_ptr neighbors; + std::shared_ptr leftNodeView; + std::shared_ptr rightNodeView; + std::shared_ptr nearbyNodeView; + std::shared_ptr randomNodeView; + std::shared_ptr handshaker; + std::shared_ptr neighborFinder; + std::shared_ptr neighborUpdateManager; + std::shared_ptr propagation; + std::shared_ptr inspector; + // Optional tuning values (TS defaults applied when absent). + std::optional neighborTargetCount; + std::optional maxContactCount; + std::optional minPropagationTargets; + std::optional maxPropagationBufferSize; + std::optional acceptProxyConnections; + std::optional neighborUpdateInterval; + std::optional rpcRequestTimeout; + bool suppressOwnMessageLoopback = false; +}; + +inline std::shared_ptr createContentDeliveryLayerNode( + ContentDeliveryLayerNodeOptions options) { + const auto ownNodeId = + Identifiers::getNodeIdFromPeerDescriptor(options.localPeerDescriptor); + const auto neighborTargetCount = + options.neighborTargetCount.value_or(defaultNeighborTargetCount); + const auto maxContactCount = + options.maxContactCount.value_or(defaultNodeViewSize); + const auto acceptProxyConnections = + options.acceptProxyConnections.value_or(defaultAcceptProxyConnections); + + auto rpcCommunicator = options.rpcCommunicator + ? options.rpcCommunicator + : std::make_shared( + formStreamPartContentDeliveryServiceId(options.streamPartId), + *options.transport); + const auto makeList = [&ownNodeId, maxContactCount]() { + return std::make_shared(ownNodeId, maxContactCount); + }; + auto neighbors = options.neighbors ? options.neighbors : makeList(); + auto leftNodeView = + options.leftNodeView ? options.leftNodeView : makeList(); + auto rightNodeView = + options.rightNodeView ? options.rightNodeView : makeList(); + auto nearbyNodeView = + options.nearbyNodeView ? options.nearbyNodeView : makeList(); + auto randomNodeView = + options.randomNodeView ? options.randomNodeView : makeList(); + auto ongoingHandshakes = std::make_shared>(); + + auto temporaryConnectionRpcLocal = + std::make_shared( + TemporaryConnectionRpcLocalOptions{ + .localPeerDescriptor = options.localPeerDescriptor, + .streamPartId = options.streamPartId, + .rpcCommunicator = *rpcCommunicator, + .connectionLocker = *options.connectionLocker}); + std::shared_ptr proxyConnectionRpcLocal; + if (acceptProxyConnections) { + proxyConnectionRpcLocal = std::make_shared( + ProxyConnectionRpcLocalOptions{ + .localPeerDescriptor = options.localPeerDescriptor, + .streamPartId = options.streamPartId, + .rpcCommunicator = *rpcCommunicator}); + } + auto propagation = options.propagation + ? options.propagation + : std::make_shared(PropagationOptions{ + .sendToNeighbor = + [neighbors, + temporaryConnectionRpcLocal, + proxyConnectionRpcLocal]( + const DhtAddress& neighborId, + const StreamMessage& msg) -> folly::coro::Task { + auto remote = neighbors->get(neighborId); + if (!remote.has_value()) { + remote = temporaryConnectionRpcLocal->getNodes().get( + neighborId); + } + if (remote.has_value()) { + co_await remote.value()->sendStreamMessage(msg); + co_return; + } + if (proxyConnectionRpcLocal) { + const auto proxyConnection = + proxyConnectionRpcLocal->getConnection(neighborId); + if (proxyConnection.has_value()) { + co_await proxyConnection.value() + ->remote.sendStreamMessage(msg); + co_return; + } + } + throw std::runtime_error("Propagation target not found"); + }, + .minPropagationTargets = options.minPropagationTargets.value_or( + defaultMinPropagationTargets), + .ttl = DEFAULT_TTL, + .maxMessages = options.maxPropagationBufferSize.value_or( + DEFAULT_MAX_MESSAGES)}); + auto handshaker = options.handshaker + ? options.handshaker + : std::make_shared(HandshakerOptions{ + .localPeerDescriptor = options.localPeerDescriptor, + .streamPartId = options.streamPartId, + .neighbors = *neighbors, + .leftNodeView = *leftNodeView, + .rightNodeView = *rightNodeView, + .nearbyNodeView = *nearbyNodeView, + .randomNodeView = *randomNodeView, + .rpcCommunicator = *rpcCommunicator, + .maxNeighborCount = neighborTargetCount, + .ongoingHandshakes = *ongoingHandshakes, + .rpcRequestTimeout = options.rpcRequestTimeout}); + auto neighborFinder = options.neighborFinder + ? options.neighborFinder + : std::static_pointer_cast( + std::make_shared(NeighborFinderOptions{ + .neighbors = *neighbors, + .nearbyNodeView = *nearbyNodeView, + .leftNodeView = *leftNodeView, + .rightNodeView = *rightNodeView, + .randomNodeView = *randomNodeView, + .doFindNeighbors = + [handshaker](std::vector excludedIds) { + return handshaker->attemptHandshakesOnContacts( + std::move(excludedIds)); + }, + .minCount = neighborTargetCount})); + auto neighborUpdateManager = options.neighborUpdateManager + ? options.neighborUpdateManager + : std::make_shared(NeighborUpdateManagerOptions{ + .localPeerDescriptor = options.localPeerDescriptor, + .streamPartId = options.streamPartId, + .neighbors = *neighbors, + .nearbyNodeView = *nearbyNodeView, + .neighborFinder = *neighborFinder, + .rpcCommunicator = *rpcCommunicator, + .neighborUpdateInterval = options.neighborUpdateInterval.value_or( + defaultNeighborUpdateInterval), + .neighborTargetCount = neighborTargetCount, + .ongoingHandshakes = *ongoingHandshakes}); + auto inspector = options.inspector + ? options.inspector + : std::make_shared(InspectorOptions{ + .localPeerDescriptor = options.localPeerDescriptor, + .streamPartId = options.streamPartId, + .rpcCommunicator = *rpcCommunicator, + .connectionLocker = *options.connectionLocker}); + + return std::make_shared( + StrictContentDeliveryLayerNodeOptions{ + .streamPartId = options.streamPartId, + .discoveryLayerNode = options.discoveryLayerNode, + .transport = options.transport, + .connectionLocker = options.connectionLocker, + .localPeerDescriptor = options.localPeerDescriptor, + .nodeViewSize = maxContactCount, + .rpcCommunicator = rpcCommunicator, + .ongoingHandshakes = ongoingHandshakes, + .nearbyNodeView = nearbyNodeView, + .randomNodeView = randomNodeView, + .leftNodeView = leftNodeView, + .rightNodeView = rightNodeView, + .neighbors = neighbors, + .temporaryConnectionRpcLocal = temporaryConnectionRpcLocal, + .proxyConnectionRpcLocal = proxyConnectionRpcLocal, + .propagation = propagation, + .handshaker = handshaker, + .neighborFinder = neighborFinder, + .neighborUpdateManager = neighborUpdateManager, + .inspector = inspector, + .neighborTargetCount = neighborTargetCount, + .isLocalNodeEntryPoint = std::move(options.isLocalNodeEntryPoint), + .rpcRequestTimeout = options.rpcRequestTimeout, + .suppressOwnMessageLoopback = options.suppressOwnMessageLoopback}); +} + +} // namespace streamr::trackerlessnetwork diff --git a/packages/streamr-trackerless-network/modules/logic/propagation/Propagation.cppm b/packages/streamr-trackerless-network/modules/logic/propagation/Propagation.cppm index 9977a6ce..2f070ec1 100644 --- a/packages/streamr-trackerless-network/modules/logic/propagation/Propagation.cppm +++ b/packages/streamr-trackerless-network/modules/logic/propagation/Propagation.cppm @@ -1,14 +1,31 @@ // Module streamr.trackerlessnetwork.Propagation // CONSOLIDATED from the former header logic/propagation/Propagation.hpp // (MODERNIZATION.md Phase 2.6): this file is now the source of truth. +// +// Async sends (phase C3): the TS sendToNeighbor is async and never +// blocks the caller; the earlier C++ port ran it synchronously, which +// meant a blockingWait inside the RPC delivery path — under broadcast +// fan-out the nodes blocked sending to each other in cycles and message +// propagation stalled (the 64-node Propagation test timeout). Sends are +// now Task-returning and run as bounded tasks on a GuardedAsyncScope; +// synchronous callers that need the outcome await collectResults(). module; +#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.Propagation; +import streamr.utils.CoroutineHelper; +import streamr.utils.GuardedAsyncScope; +import streamr.utils.SharedExecutors; import streamr.dht.Identifiers; import streamr.dht.protos; import streamr.trackerlessnetwork.PropagationTaskStore; @@ -18,11 +35,12 @@ import streamr.trackerlessnetwork.PropagationTaskStore; // differently at file scope than inside the package namespace. using streamr::dht::DhtAddress; using streamr::dht::Identifiers; +using streamr::utils::GuardedAsyncScope; export namespace streamr::trackerlessnetwork::propagation { using ::dht::PeerDescriptor; -using SendToNeighborFn = - std::function; +using SendToNeighborFn = std::function( + const DhtAddress&, const StreamMessage&)>; struct PropagationOptions { SendToNeighborFn sendToNeighbor; @@ -51,6 +69,11 @@ private: SendToNeighborFn sendToNeighbor; size_t minPropagationTargets; PropagationTaskStore activeTaskStore; + // Serializes handledNeighbors bookkeeping (sends themselves run + // concurrently and are bounded by the RPC timeouts inside + // sendToNeighbor). + std::mutex mutex; + GuardedAsyncScope scope; public: explicit Propagation(const PropagationOptions& options) @@ -61,71 +84,122 @@ public: this->minPropagationTargets = options.minPropagationTargets; } + ~Propagation() { this->stop(); } + + void stop() { this->scope.close(); } + /** - * Node should invoke this when it learns about a new message - * @return number of neighbors the message was sent to immediately + * Node should invoke this when it learns about a new message. + * Targets are node ids; sends run detached (TS parity — the caller, + * often an RPC delivery handler, must not block on them). */ - std::pair< - std::vector /* failed to send to */, - std::vector /* successfully sent to */> - feedUnseenMessage( + void feedUnseenMessage( const StreamMessage& message, - const std::vector& targets, + const std::vector& targets, const std::optional& source) { - PropagationTask task{ + auto task = std::make_shared(PropagationTask{ .message = message, .source = source, - .handledNeighbors = std::set()}; - - this->activeTaskStore.add(task); - std::vector succesfulSends; - std::vector failedSends; - for (const auto& target : targets) { - const auto neighborId = - Identifiers::getNodeIdFromPeerDescriptor(target); - if (this->sendAndAwaitThenMark(task, neighborId)) { - succesfulSends.push_back(target); + .handledNeighbors = std::set()}); + this->activeTaskStore.add(*task); + for (const auto& neighborId : targets) { + this->scheduleSend(task, neighborId); + } + } + + /** + * Awaitable variant for callers that need the per-target outcome + * (the proxy-client shared-library API): resolves once every send + * has completed, with the (failed, successful) id lists. + */ + folly::coro::Task< + std::pair, std::vector>> + feedUnseenMessageAndCollect( + StreamMessage message, + std::vector targets, + std::optional source) { + auto task = std::make_shared(PropagationTask{ + .message = std::move(message), + .source = std::move(source), + .handledNeighbors = std::set()}); + this->activeTaskStore.add(*task); + std::vector> sends; + sends.reserve(targets.size()); + for (const auto& neighborId : targets) { + sends.push_back(this->sendAndAwaitThenMark(task, neighborId)); + } + // (task is a shared_ptr; every send marks the same set.) + const auto results = + co_await folly::coro::collectAllTryRange(std::move(sends)); + std::vector failed; + std::vector successful; + for (size_t i = 0; i < results.size(); i++) { + if (results[i].hasValue() && results[i].value()) { + successful.push_back(targets[i]); } else { - failedSends.push_back(target); + failed.push_back(targets[i]); } } - return {failedSends, succesfulSends}; + co_return {failed, successful}; } /** - * Node should invoke this when it learns about a new node stream assignment + * Node should invoke this when it learns about a new node stream + * assignment. */ void onNeighborJoined(const DhtAddress& neighborId) { - const auto tasks = this->activeTaskStore.get(); - for (auto task : tasks) { - this->sendAndAwaitThenMark(task, neighborId); + // store.get() returns snapshots; each retry marks its own copy + // (mirrors the pre-existing TS-comment semantics). + for (const auto& task : this->activeTaskStore.get()) { + this->scheduleSend( + std::make_shared(task), neighborId); } } private: - bool sendAndAwaitThenMark( - PropagationTask& task, const DhtAddress& neighborId) { - if (!task.handledNeighbors.contains(neighborId) && - neighborId != task.source) { - try { - this->sendToNeighbor(neighborId, task.message); - } catch (...) { - return false; + void scheduleSend( + std::shared_ptr task, DhtAddress neighborId) { + this->scope.add( + streamr::utils::co_withExecutor( + &streamr::utils::SharedExecutors::worker(), + folly::coro::co_invoke( + [this, + task = std::move(task), + neighborId = + std::move(neighborId)]() -> folly::coro::Task { + co_await this->sendAndAwaitThenMark(task, neighborId); + }))); + } + + folly::coro::Task sendAndAwaitThenMark( + std::shared_ptr task, DhtAddress neighborId) { + { + std::scoped_lock lock(this->mutex); + if (task->handledNeighbors.contains(neighborId) || + neighborId == task->source) { + co_return false; } - // Side-note: due to asynchronicity, the task being modified at - // this point could already be stale and deleted from - // `activeTaskStore`. However, as modifying it or re-deleting it - // is pretty much inconsequential at this point, leaving the - // logic as is. - task.handledNeighbors.insert(neighborId); - if (task.handledNeighbors.size() >= this->minPropagationTargets) { + } + try { + co_await this->sendToNeighbor(neighborId, task->message); + } catch (...) { + co_return false; + } + // Side-note: due to asynchronicity, the task being modified at + // this point could already be stale and deleted from + // `activeTaskStore`. However, as modifying it or re-deleting it + // is pretty much inconsequential at this point, leaving the + // logic as is (mirrors the TS comment). + { + std::scoped_lock lock(this->mutex); + task->handledNeighbors.insert(neighborId); + if (task->handledNeighbors.size() >= this->minPropagationTargets) { this->activeTaskStore.remove( PropagationTaskStore::messageIdToMessageRef( - task.message.messageid())); + task->message.messageid())); } - return true; } - return false; + co_return true; } }; diff --git a/packages/streamr-trackerless-network/modules/logic/proxy/ProxyClient.cppm b/packages/streamr-trackerless-network/modules/logic/proxy/ProxyClient.cppm index 1004439a..16e68428 100644 --- a/packages/streamr-trackerless-network/modules/logic/proxy/ProxyClient.cppm +++ b/packages/streamr-trackerless-network/modules/logic/proxy/ProxyClient.cppm @@ -155,7 +155,9 @@ public: .localPeerDescriptor = options.localPeerDescriptor, .streamPartId = options.streamPartId, .markAndCheckDuplicate = - [this](const MessageID& msg, const MessageRef& prev) { + [this]( + const MessageID& msg, + const std::optional& prev) { return Utils::markAndCheckDuplicate( this->duplicateDetectors, msg, prev); }, @@ -184,16 +186,15 @@ public: .sendToNeighbor = [this]( const DhtAddress& neighborId, - const StreamMessage& msg) { - const auto remote = this->neighbors.get(neighborId); - if (remote.has_value()) { - streamr::utils::blockingWait( - remote.value()->sendStreamMessage(msg)); - } else { - throw std::runtime_error( - "Propagation target not found"); - } - }, + const StreamMessage& msg) -> folly::coro::Task { + const auto remote = this->neighbors.get(neighborId); + if (remote.has_value()) { + co_await remote.value()->sendStreamMessage(msg); + } else { + throw std::runtime_error( + "Propagation target not found"); + } + }, .minPropagationTargets = options.minPropagationTargets.has_value() ? options.minPropagationTargets.value() @@ -431,8 +432,23 @@ public: this->emit(msg); SLogger::debug( "ProxyClient::broadcast() calling propagation.feedUnseenMessage()"); - return this->propagation.feedUnseenMessage( - msg, this->neighbors.getPeerDescriptors(), previousNode); + // feedUnseenMessage speaks node ids (TS parity); this public API + // reports full descriptors, so map the ids back through the + // neighbor list. + const auto [failedIds, successfulIds] = streamr::utils::blockingWait( + this->propagation.feedUnseenMessageAndCollect( + msg, this->neighbors.getIds(), previousNode)); + const auto toDescriptors = [this](const std::vector& ids) { + std::vector descriptors; + for (const auto& id : ids) { + const auto remote = this->neighbors.get(id); + if (remote.has_value()) { + descriptors.push_back(remote.value()->getPeerDescriptor()); + } + } + return descriptors; + }; + return {toDescriptors(failedIds), toDescriptors(successfulIds)}; } [[nodiscard]] bool hasConnection( diff --git a/packages/streamr-trackerless-network/test/unit/ContentDeliveryLayerNodeLayer1Test.cpp b/packages/streamr-trackerless-network/test/unit/ContentDeliveryLayerNodeLayer1Test.cpp new file mode 100644 index 00000000..6b7dbba5 --- /dev/null +++ b/packages/streamr-trackerless-network/test/unit/ContentDeliveryLayerNodeLayer1Test.cpp @@ -0,0 +1,363 @@ +// Ported from packages/trackerless-network/test/integration/ +// ContentDeliveryLayerNode-Layer1Node.test.ts and its -Latencies +// variant (v103.8.0-rc.3): content delivery nodes stacked on real +// layer-1 DhtNodes over simulator transports — single node, 4 nodes +// with bidirectionality, and the 64-node scale case; the Latencies +// fixture runs the same protocol over a 50 ms fixed-latency simulator +// (the TS files differ only in that and in timeouts, so the test +// bodies are shared). +// +// 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 // IWYU pragma: keep + +import streamr.utils.CoroutineHelper; +import streamr.trackerlessnetwork.ContentDeliveryLayerNode; +import streamr.trackerlessnetwork.createContentDeliveryLayerNode; +import streamr.trackerlessnetwork.DhtNodeDiscoveryLayer; +import streamr.dht.DhtNode; +import streamr.dht.Identifiers; +import streamr.dht.Simulator; +import streamr.dht.SimulatorTransport; +import streamr.dht.protos; +import streamr.logger.SLogger; +import streamr.utils.StreamPartID; +import streamr.utils.waitForCondition; + +using ::dht::PeerDescriptor; +using streamr::dht::DhtNode; +using streamr::dht::DhtNodeOptions; +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::logger::SLogger; +using streamr::trackerlessnetwork::ContentDeliveryLayerNode; +using streamr::trackerlessnetwork::ContentDeliveryLayerNodeOptions; +using streamr::trackerlessnetwork::createContentDeliveryLayerNode; +using streamr::trackerlessnetwork::discoverylayer::DhtNodeDiscoveryLayer; +using streamr::utils::blockingWait; +using streamr::utils::StreamPartID; +using streamr::utils::StreamPartIDUtils; +using streamr::utils::waitForCondition; + +namespace { + +// Local copy of the TestUtils factory: importing the TestUtils module on +// top of this TU's DhtNode + simulator + content-delivery composition +// exhausts clang's per-TU source-location space. +inline PeerDescriptor createMockPeerDescriptor() { + PeerDescriptor descriptor; + descriptor.set_nodeid( + Identifiers::getRawFromDhtAddress( + Identifiers::createRandomDhtAddress())); + descriptor.set_type(::dht::NodeType::NODEJS); + return descriptor; +} + +constexpr size_t otherNodeCount = 64; +constexpr std::chrono::milliseconds neighborUpdateInterval{2000}; +constexpr std::chrono::seconds untilTimeout{15}; +constexpr std::chrono::seconds scaleTimeout{30}; +constexpr std::chrono::milliseconds pollInterval{100}; +constexpr size_t neighborTarget = 4; +constexpr double fixedLatencyMs = 50.0; + +// The TS createMockContentDeliveryLayerNodeAndDhtNode helper: one +// simulator transport + a layer-1 DhtNode wrapped as the discovery +// layer + a content-delivery node on top. +struct SimNode { + std::shared_ptr transport; + std::shared_ptr discoveryLayerNode; + std::shared_ptr contentDeliveryLayerNode; +}; + +SimNode createSimNode( + const PeerDescriptor& localPeerDescriptor, + const StreamPartID& streamPartId, + Simulator& simulator) { + auto transport = + std::make_shared(localPeerDescriptor, simulator); + transport->start(); + auto dhtNode = std::make_shared(DhtNodeOptions{ + .serviceId = ServiceID{streamPartId}, + .transport = transport.get(), + .connectionsView = transport.get(), + .connectionLocker = transport.get(), + .peerDescriptor = localPeerDescriptor}); + auto discoveryLayerNode = std::make_shared(dhtNode); + auto contentDeliveryLayerNode = createContentDeliveryLayerNode( + ContentDeliveryLayerNodeOptions{ + .streamPartId = streamPartId, + .discoveryLayerNode = discoveryLayerNode, + .transport = transport.get(), + .connectionLocker = transport.get(), + .localPeerDescriptor = localPeerDescriptor, + .isLocalNodeEntryPoint = []() { return false; }, + .neighborUpdateInterval = neighborUpdateInterval}); + return SimNode{ + .transport = std::move(transport), + .discoveryLayerNode = std::move(discoveryLayerNode), + .contentDeliveryLayerNode = std::move(contentDeliveryLayerNode)}; +} + +} // namespace + +class ContentDeliveryLayerNodeLayer1Test : public ::testing::Test { +protected: + StreamPartID streamPartId = StreamPartIDUtils::parse("stream#0"); + PeerDescriptor entrypointDescriptor = createMockPeerDescriptor(); + Simulator simulator; + SimNode entryPoint; + std::vector others; + + ContentDeliveryLayerNodeLayer1Test() : simulator(LatencyType::NONE) {} + ContentDeliveryLayerNodeLayer1Test( + LatencyType latencyType, std::optional fixedLatencyMs) + : simulator(latencyType, fixedLatencyMs) {} + + void SetUp() override { + this->entryPoint = createSimNode( + this->entrypointDescriptor, this->streamPartId, this->simulator); + this->others.reserve(otherNodeCount); + for (size_t i = 0; i < otherNodeCount; i++) { + this->others.push_back(createSimNode( + createMockPeerDescriptor(), + this->streamPartId, + this->simulator)); + } + blockingWait(this->entryPoint.discoveryLayerNode->start()); + blockingWait(this->entryPoint.contentDeliveryLayerNode->start()); + blockingWait(this->entryPoint.discoveryLayerNode->joinDht( + {this->entrypointDescriptor})); + { + std::vector> starts; + starts.reserve(otherNodeCount); + for (auto& other : this->others) { + starts.push_back(other.discoveryLayerNode->start()); + } + blockingWait(folly::coro::collectAllRange(std::move(starts))); + } + } + + void TearDown() override { + blockingWait(this->entryPoint.discoveryLayerNode->stop()); + this->entryPoint.contentDeliveryLayerNode->stop(); + for (auto& other : this->others) { + blockingWait(other.discoveryLayerNode->stop()); + other.contentDeliveryLayerNode->stop(); + } + this->entryPoint.transport->stop(); + for (auto& other : this->others) { + other.transport->stop(); + } + this->simulator.stop(); + } + + void runHappyPathSingleNode() { + blockingWait(this->others[0].contentDeliveryLayerNode->start()); + blockingWait(this->others[0].discoveryLayerNode->joinDht( + {this->entrypointDescriptor})); + + auto& node = *this->others[0].contentDeliveryLayerNode; + blockingWait(waitForCondition( + [&node]() { return node.getNeighbors().size() == 1; }, + untilTimeout, + pollInterval)); + EXPECT_EQ(node.getNearbyNodeView().getIds().size(), 1); + EXPECT_EQ(node.getNeighbors().size(), 1); + } + + void runHappyPathFourNodes() { + constexpr size_t nodeCount = 4; + for (size_t i = 0; i < nodeCount; i++) { + blockingWait(this->others[i].contentDeliveryLayerNode->start()); + } + { + std::vector> joins; + joins.reserve(nodeCount); + for (size_t i = 0; i < nodeCount; i++) { + joins.push_back(this->others[i].discoveryLayerNode->joinDht( + {this->entrypointDescriptor})); + } + blockingWait(folly::coro::collectAllRange(std::move(joins))); + } + blockingWait(waitForCondition( + [this]() { + for (size_t i = 0; i < nodeCount; i++) { + if (this->others[i] + .contentDeliveryLayerNode->getNeighbors() + .size() != neighborTarget) { + return false; + } + } + return true; + }, + untilTimeout, + pollInterval)); + for (size_t i = 0; i < nodeCount; i++) { + EXPECT_GE( + this->others[i] + .contentDeliveryLayerNode->getNearbyNodeView() + .getIds() + .size(), + neighborTarget); + EXPECT_GE( + this->others[i].contentDeliveryLayerNode->getNeighbors().size(), + neighborTarget); + } + + // Check bidirectionality across the four nodes + the entry point. + std::vector allNodes; + allNodes.reserve(nodeCount + 1); + for (size_t i = 0; i < nodeCount; i++) { + allNodes.push_back(this->others[i].contentDeliveryLayerNode.get()); + } + allNodes.push_back(this->entryPoint.contentDeliveryLayerNode.get()); + for (auto* node : allNodes) { + for (const auto& nodeId : node->getNearbyNodeView().getIds()) { + const auto neighborIt = std::ranges::find_if( + allNodes, [&nodeId](ContentDeliveryLayerNode* candidate) { + return candidate->getOwnNodeId() == nodeId; + }); + ASSERT_NE(neighborIt, allNodes.end()); + const auto neighborsOfNeighbor = (*neighborIt)->getNeighbors(); + const auto includesNode = std::ranges::any_of( + neighborsOfNeighbor, + [node](const PeerDescriptor& descriptor) { + return Identifiers::getNodeIdFromPeerDescriptor( + descriptor) == node->getOwnNodeId(); + }); + EXPECT_EQ(includesNode, true); + } + } + } + + void runHappyPathSixtyFourNodes() { + for (auto& other : this->others) { + blockingWait(other.contentDeliveryLayerNode->start()); + } + { + std::vector> joins; + joins.reserve(otherNodeCount); + for (auto& other : this->others) { + joins.push_back(other.discoveryLayerNode->joinDht( + {this->entrypointDescriptor})); + } + blockingWait(folly::coro::collectAllRange(std::move(joins))); + } + blockingWait(waitForCondition( + [this]() { + return std::ranges::all_of(this->others, [](const auto& other) { + return other.contentDeliveryLayerNode->getNeighbors() + .size() >= neighborTarget; + }); + }, + scaleTimeout, + pollInterval)); + + size_t neighborSum = 0; + for (const auto& other : this->others) { + neighborSum += + other.contentDeliveryLayerNode->getNeighbors().size(); + } + SLogger::info( + "AVG number of neighbors: " + + std::to_string(static_cast(neighborSum) / otherNodeCount)); + + blockingWait(waitForCondition( + [this]() { + return std::ranges::all_of(this->others, [](const auto& other) { + return other.contentDeliveryLayerNode + ->getOutgoingHandshakeCount() == 0; + }); + }, + scaleTimeout, + pollInterval)); + + // TS NET-1074: sometimes unidirectional connections remain — accept + // up to 2 mismatches. + blockingWait(waitForCondition( + [this]() { + size_t mismatchCounter = 0; + for (const auto& other : this->others) { + const auto nodeId = + other.contentDeliveryLayerNode->getOwnNodeId(); + for (const auto& neighbor : + other.contentDeliveryLayerNode->getNeighbors()) { + const auto neighborId = + Identifiers::getNodeIdFromPeerDescriptor(neighbor); + if (neighborId == + this->entryPoint.contentDeliveryLayerNode + ->getOwnNodeId()) { + continue; + } + const auto neighborIt = std::ranges::find_if( + this->others, [&neighborId](const auto& candidate) { + return candidate.contentDeliveryLayerNode + ->getOwnNodeId() == neighborId; + }); + if (neighborIt == this->others.end()) { + continue; + } + const auto neighborsOfNeighbor = + neighborIt->contentDeliveryLayerNode + ->getNeighbors(); + const auto includesNode = std::ranges::any_of( + neighborsOfNeighbor, + [&nodeId](const PeerDescriptor& descriptor) { + return Identifiers::getNodeIdFromPeerDescriptor( + descriptor) == nodeId; + }); + if (!includesNode) { + mismatchCounter++; + } + } + } + return mismatchCounter <= 2; + }, + scaleTimeout, + std::chrono::seconds(1))); + } +}; + +// The -Latencies TS variant: same protocol over 50 ms fixed latency. +class ContentDeliveryLayerNodeLayer1LatenciesTest + : public ContentDeliveryLayerNodeLayer1Test { +protected: + ContentDeliveryLayerNodeLayer1LatenciesTest() + : ContentDeliveryLayerNodeLayer1Test( + LatencyType::FIXED, fixedLatencyMs) {} +}; + +TEST_F(ContentDeliveryLayerNodeLayer1Test, HappyPathSingleNode) { + this->runHappyPathSingleNode(); +} + +TEST_F(ContentDeliveryLayerNodeLayer1Test, HappyPathFourNodes) { + this->runHappyPathFourNodes(); +} + +TEST_F(ContentDeliveryLayerNodeLayer1Test, HappyPathSixtyFourNodes) { + this->runHappyPathSixtyFourNodes(); +} + +TEST_F(ContentDeliveryLayerNodeLayer1LatenciesTest, HappyPathSingleNode) { + this->runHappyPathSingleNode(); +} + +TEST_F(ContentDeliveryLayerNodeLayer1LatenciesTest, HappyPathFourNodes) { + this->runHappyPathFourNodes(); +} + +TEST_F(ContentDeliveryLayerNodeLayer1LatenciesTest, HappyPathSixtyFourNodes) { + this->runHappyPathSixtyFourNodes(); +} diff --git a/packages/streamr-trackerless-network/test/unit/ContentDeliveryLayerNodeTest.cpp b/packages/streamr-trackerless-network/test/unit/ContentDeliveryLayerNodeTest.cpp new file mode 100644 index 00000000..db61087a --- /dev/null +++ b/packages/streamr-trackerless-network/test/unit/ContentDeliveryLayerNodeTest.cpp @@ -0,0 +1,270 @@ +// Ported from packages/trackerless-network/test/unit/ +// ContentDeliveryLayerNode.test.ts (v103.8.0-rc.3): view updates driven +// by discovery-layer events, neighbor info, and the +// suppressOwnMessageLoopback delivery flag. +// +// Adaptations: the TS test injects MockHandshaker / +// MockNeighborUpdateManager "as any"; here the factory builds the real +// components, which stay inert within the test's lifetime (the update +// interval is 10 s and no handshakes are triggered). The getInfos +// assertions match by descriptor rather than index because the C++ +// NodeList is id-ordered, not insertion-ordered. +// +// 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.trackerlessnetwork.ContentDeliveryLayerNode; +import streamr.trackerlessnetwork.createContentDeliveryLayerNode; +import streamr.trackerlessnetwork.DiscoveryLayerNode; +import streamr.trackerlessnetwork.MockDiscoveryLayerNode; +import streamr.trackerlessnetwork.NodeList; +import streamr.trackerlessnetwork.TestUtils; +import streamr.trackerlessnetwork.protos; +import streamr.dht.FakeTransport; +import streamr.dht.Identifiers; +import streamr.dht.protos; +import streamr.utils.EthereumAddress; +import streamr.utils.StreamPartID; +import streamr.utils.waitForCondition; + +using ::dht::PeerDescriptor; +using streamr::dht::Identifiers; +using streamr::dht::transport::FakeTransport; +using streamr::trackerlessnetwork::ContentDeliveryLayerNode; +using streamr::trackerlessnetwork::ContentDeliveryLayerNodeOptions; +using streamr::trackerlessnetwork::createContentDeliveryLayerNode; +using streamr::trackerlessnetwork::NodeList; +using streamr::trackerlessnetwork::contentdeliverylayernodeevents::Message; +using streamr::trackerlessnetwork::testutils:: + createMockContentDeliveryRpcRemote; +using streamr::trackerlessnetwork::testutils::createMockPeerDescriptor; +using streamr::trackerlessnetwork::testutils::createStreamMessage; +using streamr::trackerlessnetwork::testutils::MockConnectionLocker; +using streamr::trackerlessnetwork::testutils::MockDiscoveryLayerNode; +using streamr::utils::blockingWait; +using streamr::utils::StreamPartIDUtils; +using streamr::utils::toEthereumAddress; +using streamr::utils::waitForCondition; + +namespace dle = + streamr::trackerlessnetwork::discoverylayer::discoverylayernodeevents; + +namespace { +constexpr size_t viewLimit = 10; +constexpr std::chrono::seconds untilTimeout{10}; +constexpr std::chrono::milliseconds untilPollInterval{100}; +} // namespace + +class ContentDeliveryLayerNodeTest : public ::testing::Test { +protected: + PeerDescriptor peerDescriptor = createMockPeerDescriptor(); + FakeTransport transport{peerDescriptor, [](const auto& /*message*/) {}}; + MockConnectionLocker connectionLocker; + std::shared_ptr neighbors; + std::shared_ptr nearbyNodeView; + std::shared_ptr randomNodeView; + std::shared_ptr discoveryLayerNode; + std::shared_ptr node; + + void SetUp() override { + const auto nodeId = + Identifiers::getNodeIdFromPeerDescriptor(this->peerDescriptor); + this->neighbors = std::make_shared(nodeId, viewLimit); + this->randomNodeView = std::make_shared(nodeId, viewLimit); + this->nearbyNodeView = std::make_shared(nodeId, viewLimit); + this->discoveryLayerNode = std::make_shared(); + this->node = this->buildNode(false); + blockingWait(this->node->start()); + } + + void TearDown() override { this->node->stop(); } + + std::shared_ptr buildNode( + bool suppressOwnMessageLoopback) { + return createContentDeliveryLayerNode( + ContentDeliveryLayerNodeOptions{ + .streamPartId = StreamPartIDUtils::parse("stream#0"), + .discoveryLayerNode = this->discoveryLayerNode, + .transport = &this->transport, + .connectionLocker = &this->connectionLocker, + .localPeerDescriptor = this->peerDescriptor, + .isLocalNodeEntryPoint = []() { return false; }, + .neighbors = this->neighbors, + .nearbyNodeView = this->nearbyNodeView, + .randomNodeView = this->randomNodeView, + .suppressOwnMessageLoopback = suppressOwnMessageLoopback}); + } + + // Self-contained node for the suppressOwnMessageLoopback tests (the + // TS test builds fresh views/transport per node too). + struct StandaloneNode { + PeerDescriptor peerDescriptor = createMockPeerDescriptor(); + FakeTransport transport{peerDescriptor, [](const auto& /*message*/) {}}; + MockConnectionLocker connectionLocker; + std::shared_ptr discoveryLayerNode = + std::make_shared(); + std::shared_ptr node; + + explicit StandaloneNode(bool suppressOwnMessageLoopback) { + this->node = createContentDeliveryLayerNode( + ContentDeliveryLayerNodeOptions{ + .streamPartId = StreamPartIDUtils::parse("stream#0"), + .discoveryLayerNode = this->discoveryLayerNode, + .transport = &this->transport, + .connectionLocker = &this->connectionLocker, + .localPeerDescriptor = this->peerDescriptor, + .isLocalNodeEntryPoint = []() { return false; }, + .suppressOwnMessageLoopback = suppressOwnMessageLoopback}); + } + + ~StandaloneNode() { this->node->stop(); } + }; + + static StreamMessage ownMessage() { + return createStreamMessage( + "x", + StreamPartIDUtils::parse("stream#0"), + toEthereumAddress("0x1234567890123456789012345678901234567890")); + } +}; + +TEST_F(ContentDeliveryLayerNodeTest, GetNeighbors) { + const auto mockRemote = createMockContentDeliveryRpcRemote(); + this->neighbors->add(mockRemote); + const auto result = this->node->getNeighbors(); + ASSERT_EQ(result.size(), 1); + EXPECT_EQ( + Identifiers::getNodeIdFromPeerDescriptor(result[0]), + Identifiers::getNodeIdFromPeerDescriptor( + mockRemote->getPeerDescriptor())); +} + +TEST_F(ContentDeliveryLayerNodeTest, GetNearbyNodeView) { + const auto mockRemote = createMockContentDeliveryRpcRemote(); + this->nearbyNodeView->add(mockRemote); + const auto ids = this->node->getNearbyNodeView().getIds(); + ASSERT_EQ(ids.size(), 1); + EXPECT_EQ( + ids[0], + Identifiers::getNodeIdFromPeerDescriptor( + mockRemote->getPeerDescriptor())); +} + +TEST_F(ContentDeliveryLayerNodeTest, AddsClosestNodesToNearbyNodeView) { + const auto peerDescriptor1 = createMockPeerDescriptor(); + const auto peerDescriptor2 = createMockPeerDescriptor(); + this->discoveryLayerNode->setClosestContacts( + {peerDescriptor1, peerDescriptor2}); + this->discoveryLayerNode->emit(peerDescriptor1); + blockingWait(waitForCondition( + [this]() { return this->nearbyNodeView->size() == 2; }, + untilTimeout, + untilPollInterval)); + EXPECT_TRUE( + this->nearbyNodeView + ->get(Identifiers::getNodeIdFromPeerDescriptor(peerDescriptor1)) + .has_value()); + EXPECT_TRUE( + this->nearbyNodeView + ->get(Identifiers::getNodeIdFromPeerDescriptor(peerDescriptor2)) + .has_value()); +} + +TEST_F(ContentDeliveryLayerNodeTest, AddsRandomNodesToRandomNodeView) { + const auto peerDescriptor1 = createMockPeerDescriptor(); + const auto peerDescriptor2 = createMockPeerDescriptor(); + this->discoveryLayerNode->setRandomContacts( + {peerDescriptor1, peerDescriptor2}); + this->discoveryLayerNode->emit(peerDescriptor1); + blockingWait(waitForCondition( + [this]() { return this->randomNodeView->size() == 2; }, + untilTimeout, + untilPollInterval)); + EXPECT_TRUE( + this->randomNodeView + ->get(Identifiers::getNodeIdFromPeerDescriptor(peerDescriptor1)) + .has_value()); + EXPECT_TRUE( + this->randomNodeView + ->get(Identifiers::getNodeIdFromPeerDescriptor(peerDescriptor2)) + .has_value()); +} + +TEST_F(ContentDeliveryLayerNodeTest, AddsDiscoveryNeighborsToNearbyNodeView) { + const auto peerDescriptor1 = createMockPeerDescriptor(); + const auto peerDescriptor2 = createMockPeerDescriptor(); + this->discoveryLayerNode->addNewRandomPeerToKBucket(); + this->discoveryLayerNode->setClosestContacts( + {peerDescriptor1, peerDescriptor2}); + this->discoveryLayerNode->emit(peerDescriptor1); + blockingWait(waitForCondition( + [this]() { return this->nearbyNodeView->size() == 3; }, + untilTimeout, + untilPollInterval)); + EXPECT_TRUE( + this->nearbyNodeView + ->get(Identifiers::getNodeIdFromPeerDescriptor(peerDescriptor1)) + .has_value()); + EXPECT_TRUE( + this->nearbyNodeView + ->get(Identifiers::getNodeIdFromPeerDescriptor(peerDescriptor2)) + .has_value()); +} + +TEST_F(ContentDeliveryLayerNodeTest, GetInfos) { + const auto nodeWithRtt = createMockContentDeliveryRpcRemote(); + this->neighbors->add(nodeWithRtt); + const auto nodeWithoutRtt = createMockContentDeliveryRpcRemote(); + this->neighbors->add(nodeWithoutRtt); + nodeWithRtt->setRtt(100); // NOLINT + const auto infos = this->node->getInfos(); + ASSERT_EQ(infos.size(), 2); + const auto rttNodeId = Identifiers::getNodeIdFromPeerDescriptor( + nodeWithRtt->getPeerDescriptor()); + for (const auto& info : infos) { + if (Identifiers::getNodeIdFromPeerDescriptor(info.peerDescriptor) == + rttNodeId) { + EXPECT_EQ(info.rtt, 100); + } else { + EXPECT_EQ(info.rtt.has_value(), false); + } + } +} + +TEST_F(ContentDeliveryLayerNodeTest, OwnPublishDeliveredByDefault) { + // The fixture node has suppressOwnMessageLoopback = false. + size_t received = 0; + this->node->on( + [&received](const StreamMessage& /*msg*/) { received++; }); + this->node->broadcast(this->ownMessage()); // no previousNode + EXPECT_EQ(received, 1); +} + +TEST_F(ContentDeliveryLayerNodeTest, SuppressedOwnPublishNotDelivered) { + StandaloneNode standalone{true}; + blockingWait(standalone.node->start()); + size_t received = 0; + standalone.node->on( + [&received](const StreamMessage& /*msg*/) { received++; }); + standalone.node->broadcast(this->ownMessage()); // no previousNode + EXPECT_EQ(received, 0); +} + +TEST_F(ContentDeliveryLayerNodeTest, SuppressedForwardedMessageIsDelivered) { + StandaloneNode standalone{true}; + blockingWait(standalone.node->start()); + size_t received = 0; + standalone.node->on( + [&received](const StreamMessage& /*msg*/) { received++; }); + standalone.node->broadcast( + this->ownMessage(), Identifiers::createRandomDhtAddress()); + EXPECT_EQ(received, 1); +} diff --git a/packages/streamr-trackerless-network/test/unit/NodeListTest.cpp b/packages/streamr-trackerless-network/test/unit/NodeListTest.cpp index c9c19838..24946f2f 100644 --- a/packages/streamr-trackerless-network/test/unit/NodeListTest.cpp +++ b/packages/streamr-trackerless-network/test/unit/NodeListTest.cpp @@ -1,9 +1,188 @@ +// Ported from packages/trackerless-network/test/unit/NodeList.test.ts +// (v103.8.0-rc.3): add/remove with the size limit, the first/last/random +// accessors with excludes, and the insertion-order contract the +// handshake-interleave protocol depends on. +#include +#include +#include +#include #include +#include "packages/dht/protos/DhtRpc.pb.h" +// The TS test drives ordering with literal 3-byte node ids; the byte +// patterns are data, not tunables. +// NOLINTBEGIN(readability-magic-numbers) + +import streamr.trackerlessnetwork.ContentDeliveryRpcRemote; import streamr.trackerlessnetwork.NodeList; +import streamr.trackerlessnetwork.TestUtils; +import streamr.dht.Identifiers; +import streamr.dht.protos; + +using ::dht::PeerDescriptor; +using streamr::dht::DhtAddress; +using streamr::dht::DhtAddressRaw; +using streamr::dht::Identifiers; +using streamr::trackerlessnetwork::ContentDeliveryRpcRemote; +using streamr::trackerlessnetwork::NodeList; +using streamr::trackerlessnetwork::testutils:: + createMockContentDeliveryRpcRemote; +using streamr::trackerlessnetwork::testutils::createMockPeerDescriptor; + +namespace { + +constexpr size_t nodeListLimit = 6; + +DhtAddress addressFromBytes(std::initializer_list bytes) { + return Identifiers::getDhtAddressFromRaw( + DhtAddressRaw{std::string{bytes.begin(), bytes.end()}}); +} + +PeerDescriptor createDescriptorWithId(const DhtAddress& nodeId) { + PeerDescriptor descriptor; + descriptor.set_nodeid(Identifiers::getRawFromDhtAddress(nodeId)); + descriptor.set_type(::dht::NodeType::NODEJS); + return descriptor; +} + +} // namespace + +class NodeListTest : public ::testing::Test { +protected: + std::vector ids = { + addressFromBytes({1, 1, 1}), + addressFromBytes({1, 1, 2}), + addressFromBytes({1, 1, 3}), + addressFromBytes({1, 1, 4}), + addressFromBytes({1, 1, 5})}; + DhtAddress ownId = Identifiers::createRandomDhtAddress(); + std::optional nodeList; -using streamr::trackerlessnetwork::NodeList; // NOLINT + void SetUp() override { + this->nodeList.emplace(this->ownId, nodeListLimit); + for (const auto& id : this->ids) { + this->nodeList->add( + createMockContentDeliveryRpcRemote(createDescriptorWithId(id))); + } + } + + [[nodiscard]] static DhtAddress nodeIdOf( + const std::shared_ptr& remote) { + return Identifiers::getNodeIdFromPeerDescriptor( + remote->getPeerDescriptor()); + } +}; + +TEST_F(NodeListTest, Add) { + const auto newDescriptor = createMockPeerDescriptor(); + this->nodeList->add(createMockContentDeliveryRpcRemote(newDescriptor)); + EXPECT_EQ( + this->nodeList->has( + Identifiers::getNodeIdFromPeerDescriptor(newDescriptor)), + true); + + // The list is at its limit now; further adds are rejected. + const auto newDescriptor2 = createMockPeerDescriptor(); + this->nodeList->add(createMockContentDeliveryRpcRemote(newDescriptor2)); + EXPECT_EQ( + this->nodeList->has( + Identifiers::getNodeIdFromPeerDescriptor(newDescriptor2)), + false); +} + +TEST_F(NodeListTest, Remove) { + const auto toRemove = this->nodeList->getFirst({}); + const auto nodeId = nodeIdOf(toRemove.value()); + this->nodeList->remove(nodeId); + EXPECT_EQ(this->nodeList->has(nodeId), false); +} + +TEST_F(NodeListTest, GetFirst) { + const auto closest = this->nodeList->getFirst({}); + EXPECT_EQ(nodeIdOf(closest.value()), addressFromBytes({1, 1, 1})); +} + +TEST_F(NodeListTest, GetFirstWithExclude) { + const auto closest = + this->nodeList->getFirst({addressFromBytes({1, 1, 1})}); + EXPECT_EQ(nodeIdOf(closest.value()), addressFromBytes({1, 1, 2})); +} -TEST(NodeListTest, ItCanBeInstantiated) { - // NodeList nodeList; +TEST_F(NodeListTest, GetFirstWsOnly) { + auto descriptor = createMockPeerDescriptor(); + descriptor.mutable_websocket()->set_port(111); // NOLINT + descriptor.mutable_websocket()->set_host(""); + descriptor.mutable_websocket()->set_tls(false); + this->nodeList->add(createMockContentDeliveryRpcRemote(descriptor)); + const auto closest = this->nodeList->getFirst({}, true); + EXPECT_EQ(closest.has_value(), true); } + +TEST_F(NodeListTest, GetLast) { + const auto last = this->nodeList->getLast({}); + EXPECT_EQ(nodeIdOf(last.value()), addressFromBytes({1, 1, 5})); +} + +TEST_F(NodeListTest, GetLastWithExclude) { + const auto last = this->nodeList->getLast({addressFromBytes({1, 1, 5})}); + EXPECT_EQ(nodeIdOf(last.value()), addressFromBytes({1, 1, 4})); +} + +TEST_F(NodeListTest, GetFirstAndLast) { + const auto results = this->nodeList->getFirstAndLast({}); + ASSERT_EQ(results.size(), 2); + EXPECT_EQ(results[0], this->nodeList->getFirst({}).value()); + EXPECT_EQ(results[1], this->nodeList->getLast({}).value()); +} + +TEST_F(NodeListTest, GetFirstEmpty) { + NodeList emptyList{this->ownId, 2}; + EXPECT_EQ(emptyList.getFirst({}).has_value(), false); +} + +TEST_F(NodeListTest, GetLastEmpty) { + NodeList emptyList{this->ownId, 2}; + EXPECT_EQ(emptyList.getLast({}).has_value(), false); +} + +TEST_F(NodeListTest, GetRandomEmpty) { + NodeList emptyList{this->ownId, 2}; + EXPECT_EQ(emptyList.getRandom({}).has_value(), false); +} + +TEST_F(NodeListTest, GetFirstAndLastEmpty) { + NodeList emptyList{this->ownId, 2}; + EXPECT_EQ(emptyList.getFirstAndLast({}).size(), 0); +} + +TEST_F(NodeListTest, GetFirstAndLastWithExclude) { + const auto results = this->nodeList->getFirstAndLast( + {addressFromBytes({1, 1, 1}), addressFromBytes({1, 1, 5})}); + ASSERT_EQ(results.size(), 2); + EXPECT_EQ( + results[0], + this->nodeList->getFirst({addressFromBytes({1, 1, 1})}).value()); + EXPECT_EQ( + results[1], + this->nodeList->getLast({addressFromBytes({1, 1, 5})}).value()); +} + +TEST_F(NodeListTest, ItemsAreInInsertionOrder) { + NodeList list{Identifiers::createRandomDhtAddress(), 100}; // NOLINT + const auto item1 = createMockContentDeliveryRpcRemote(); + const auto item2 = createMockContentDeliveryRpcRemote(); + const auto item3 = createMockContentDeliveryRpcRemote(); + const auto item4 = createMockContentDeliveryRpcRemote(); + const auto item5 = createMockContentDeliveryRpcRemote(); + const auto item6 = createMockContentDeliveryRpcRemote(); + list.add(item2); + list.add(item3); + list.add(item1); + list.add(item6); + list.add(item4); + list.add(item5); + EXPECT_EQ(list.getFirst({}).value(), item2); + EXPECT_EQ(list.getLast({}).value(), item5); +} + +// NOLINTEND(readability-magic-numbers) diff --git a/packages/streamr-trackerless-network/test/unit/PropagationScaleTest.cpp b/packages/streamr-trackerless-network/test/unit/PropagationScaleTest.cpp new file mode 100644 index 00000000..d4c38d85 --- /dev/null +++ b/packages/streamr-trackerless-network/test/unit/PropagationScaleTest.cpp @@ -0,0 +1,226 @@ +// Ported from packages/trackerless-network/test/integration/ +// Propagation.test.ts (v103.8.0-rc.3): a broadcast from one node +// reaches every node of the stream-part overlay. +// +// Adaptation: the TS test runs 256 nodes; the C++ simulator carries a +// far higher per-node cost (two full DHT layers of real coroutines per +// node), and 256 nodes takes ~9-16 min locally — beyond the CI budget. +// 64 nodes exercises the same propagation logic over a real multi-hop +// mesh. Scaling this back to 256 (and the teardown stress it implies) +// is a documented follow-up. +// +// 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 +#include + +#include // IWYU pragma: keep + +import streamr.utils.CoroutineHelper; +import streamr.trackerlessnetwork.ContentDeliveryLayerNode; +import streamr.trackerlessnetwork.createContentDeliveryLayerNode; +import streamr.trackerlessnetwork.DhtNodeDiscoveryLayer; +import streamr.trackerlessnetwork.protos; +import streamr.dht.DhtNode; +import streamr.dht.Simulator; +import streamr.dht.SimulatorTransport; +import streamr.dht.protos; +import streamr.dht.Identifiers; +import streamr.utils.BinaryUtils; +import streamr.utils.StreamPartID; +import streamr.utils.waitForCondition; + +using ::dht::PeerDescriptor; +using streamr::dht::DhtNode; +using streamr::dht::DhtNodeOptions; +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::trackerlessnetwork::ContentDeliveryLayerNode; +using streamr::trackerlessnetwork::ContentDeliveryLayerNodeOptions; +using streamr::trackerlessnetwork::createContentDeliveryLayerNode; +using streamr::trackerlessnetwork::contentdeliverylayernodeevents::Message; +using streamr::trackerlessnetwork::discoverylayer::DhtNodeDiscoveryLayer; +using streamr::utils::BinaryUtils; +using streamr::utils::blockingWait; +using streamr::utils::StreamPartID; +using streamr::utils::StreamPartIDUtils; +using streamr::utils::waitForCondition; + +namespace { + +// Local copies of the TestUtils factories: importing the TestUtils module +// on top of this TU's DhtNode + simulator + content-delivery composition +// exhausts clang's per-TU source-location space. +inline PeerDescriptor createMockPeerDescriptor() { + PeerDescriptor descriptor; + descriptor.set_nodeid( + Identifiers::getRawFromDhtAddress( + Identifiers::createRandomDhtAddress())); + descriptor.set_type(::dht::NodeType::NODEJS); + return descriptor; +} + +inline StreamMessage createLocalStreamMessage( + const std::string& content, const StreamPartID& streamPartId) { + StreamMessage msg; + auto* messageId = msg.mutable_messageid(); + messageId->set_streamid(StreamPartIDUtils::getStreamID(streamPartId)); + messageId->set_streampartition( + static_cast( + StreamPartIDUtils::getStreamPartition(streamPartId).value_or(0))); + messageId->set_sequencenumber(0); + messageId->set_timestamp(0); + messageId->set_publisherid( + BinaryUtils::hexToBinaryString( + "0x1234567890123456789012345678901234567890")); + messageId->set_messagechainid("messageChain0"); + msg.set_signaturetype(SignatureType::ECDSA_SECP256K1_EVM); + msg.set_signature(BinaryUtils::hexToBinaryString("0x1234")); + auto* contentMessage = msg.mutable_contentmessage(); + contentMessage->set_encryptiontype(EncryptionType::NONE); + contentMessage->set_contenttype(ContentType::JSON); + contentMessage->set_content(content); + return msg; +} + +constexpr size_t nodeCount = 64; +constexpr std::chrono::seconds meshTimeout{60}; +constexpr std::chrono::seconds propagationTimeout{10}; +constexpr std::chrono::milliseconds pollInterval{200}; +// TS createMockContentDeliveryLayerNodeAndDhtNode DhtNode options. +constexpr size_t tsNumberOfNodesPerKBucket = 4; +constexpr size_t tsNeighborPingLimit = 16; +constexpr std::chrono::milliseconds tsRpcRequestTimeout{5000}; +constexpr double averageNeighborTarget = 4.0; + +struct SimNode { + std::shared_ptr transport; + std::shared_ptr discoveryLayerNode; + std::shared_ptr contentDeliveryLayerNode; +}; + +SimNode createSimNode( + const PeerDescriptor& localPeerDescriptor, + const StreamPartID& streamPartId, + Simulator& simulator) { + auto transport = + std::make_shared(localPeerDescriptor, simulator); + transport->start(); + auto dhtNode = std::make_shared(DhtNodeOptions{ + .serviceId = ServiceID{streamPartId}, + .numberOfNodesPerKBucket = tsNumberOfNodesPerKBucket, + .neighborPingLimit = tsNeighborPingLimit, + .rpcRequestTimeout = tsRpcRequestTimeout, + .transport = transport.get(), + .connectionsView = transport.get(), + .connectionLocker = transport.get(), + .peerDescriptor = localPeerDescriptor}); + auto discoveryLayerNode = std::make_shared(dhtNode); + auto contentDeliveryLayerNode = createContentDeliveryLayerNode( + ContentDeliveryLayerNodeOptions{ + .streamPartId = streamPartId, + .discoveryLayerNode = discoveryLayerNode, + .transport = transport.get(), + .connectionLocker = transport.get(), + .localPeerDescriptor = localPeerDescriptor, + .isLocalNodeEntryPoint = []() { return false; }}); + return SimNode{ + .transport = std::move(transport), + .discoveryLayerNode = std::move(discoveryLayerNode), + .contentDeliveryLayerNode = std::move(contentDeliveryLayerNode)}; +} + +} // namespace + +class PropagationScaleTest : public ::testing::Test { +protected: + StreamPartID streamPartId = StreamPartIDUtils::parse("testingtesting#0"); + PeerDescriptor entryPointDescriptor = createMockPeerDescriptor(); + Simulator simulator{LatencyType::NONE}; + std::vector nodes; + std::atomic totalReceived = 0; + + void SetUp() override { + auto entryPoint = createSimNode( + this->entryPointDescriptor, this->streamPartId, this->simulator); + blockingWait(entryPoint.discoveryLayerNode->start()); + blockingWait(entryPoint.discoveryLayerNode->joinDht( + {this->entryPointDescriptor})); + blockingWait(entryPoint.contentDeliveryLayerNode->start()); + entryPoint.contentDeliveryLayerNode->on( + [this](const StreamMessage& /*msg*/) { this->totalReceived++; }); + this->nodes.push_back(std::move(entryPoint)); + + std::vector> joins; + joins.reserve(nodeCount); + for (size_t i = 0; i < nodeCount; i++) { + auto node = createSimNode( + createMockPeerDescriptor(), + this->streamPartId, + this->simulator); + blockingWait(node.discoveryLayerNode->start()); + blockingWait(node.contentDeliveryLayerNode->start()); + node.contentDeliveryLayerNode->on( + [this](const StreamMessage& /*msg*/) { + this->totalReceived++; + }); + joins.push_back( + node.discoveryLayerNode->joinDht({this->entryPointDescriptor})); + this->nodes.push_back(std::move(node)); + } + blockingWait(folly::coro::collectAllRange(std::move(joins))); + } + + void TearDown() override { + for (auto& node : this->nodes) { + node.contentDeliveryLayerNode->stop(); + } + for (auto& node : this->nodes) { + blockingWait(node.discoveryLayerNode->stop()); + } + for (auto& node : this->nodes) { + node.transport->stop(); + } + this->simulator.stop(); + } +}; + +TEST_F(PropagationScaleTest, AllNodesReceiveMessages) { + blockingWait(waitForCondition( + [this]() { + return std::ranges::all_of(this->nodes, [](const auto& node) { + return node.contentDeliveryLayerNode->getNeighbors().size() >= + 3; + }); + }, + meshTimeout, + pollInterval)); + blockingWait(waitForCondition( + [this]() { + size_t sum = 0; + for (const auto& node : this->nodes) { + sum += node.contentDeliveryLayerNode->getNeighbors().size(); + } + return (static_cast(sum) / + static_cast(this->nodes.size())) >= + averageNeighborTarget; + }, + meshTimeout, + pollInterval)); + const auto msg = + createLocalStreamMessage(R"({"hello":"WORLD"})", this->streamPartId); + this->nodes[0].contentDeliveryLayerNode->broadcast(msg); + blockingWait(waitForCondition( + [this]() { return this->totalReceived >= nodeCount; }, + propagationTimeout, + pollInterval)); +} diff --git a/packages/streamr-trackerless-network/test/utils/TestUtils.cppm b/packages/streamr-trackerless-network/test/utils/TestUtils.cppm index 1b104633..052f81e6 100644 --- a/packages/streamr-trackerless-network/test/utils/TestUtils.cppm +++ b/packages/streamr-trackerless-network/test/utils/TestUtils.cppm @@ -7,17 +7,23 @@ // (trackerless-network-completion-plan.md, phase 0.2). 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.TestUtils; +import streamr.utils.CoroutineHelper; import streamr.protorpc.RpcCommunicator; import streamr.protorpc.protos; import streamr.trackerlessnetwork.ContentDeliveryRpcRemote;