From a78f0baaa4b9ec049de55147e0209cf7130cb2a2 Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Sun, 12 Jul 2026 20:10:53 +0300 Subject: [PATCH] Phase C5: ContentDeliveryManager and the control-layer facade Ports packages/trackerless-network/src/ContentDeliveryManager.ts and src/control-layer/ControlLayerNode.ts (TS pin af966cf03, v103.8.0-rc.3): - ContentDeliveryManager: per-stream-part join/leave/broadcast orchestration; builds the discovery-layer node, content-delivery node, entry-point store manager, split avoidance and reconnect per joined stream part, and the ProxyClient for proxied stream parts. The TS fire-and-forget setImmediate() join runs as a bounded GuardedAsyncScope task. - ControlLayerNode: the layer-0 facade (pure virtual); asTransport() replaces the TS structural "extends ITransport". - DhtNodeControlLayer: adapter presenting a DhtNode as a ControlLayerNode (mirrors DhtNodeDiscoveryLayer). - createStreamPartDiscoveryLayerNode: the TS inline layer-1 DhtNode construction as its own module; the manager takes it as an injected factory because composing the DhtNode module graph inside ContentDeliveryManager.cppm exhausts clang's per-TU source locations. - DhtNode::deleteDataFromDht (was still missing; TS parity). - Source-location relief across the package: 23 modules stop textually including DhtRpc.pb.h/NetworkRpc.pb.h in their global module fragments and rely on streamr.dht.protos / streamr.trackerlessnetwork.protos imports instead, with the std headers they actually use spelled out. Without this the new manager test TU cannot compile ("ran out of source locations"). - ContentDeliveryManagerTest: 6 tests over 2 managers on layer-0 DhtNodes over the simulator (join, pub/sub after neighbors, multi-stream, leave, RTT collection in getNodeInfo). Suite exits cleanly (no leaked executor keep-alives). - lint.sh: ContentDeliveryManagerTest.cpp joins the clangd-tidy std-type false-positive exclusion list (generated protobuf setter; the compiler builds and runs it). Deviations documented in the module header: TS metrics/diagnostics not ported (consistent with earlier phases), ConnectionManager private-client mode is a follow-up, plumtree options are milestone E. The TS proxy e2e tests target the C6 NetworkNode API and move there. Co-Authored-By: Claude Fable 5 --- packages/streamr-dht/modules/dht/DhtNode.cppm | 12 + .../CMakeLists.txt | 1 + packages/streamr-trackerless-network/lint.sh | 6 +- .../modules/ContentDeliveryManager.cppm | 642 ++++++++++++++++++ .../StreamPartNetworkSplitAvoidance.cppm | 1 + .../control-layer/ControlLayerNode.cppm | 58 ++ .../control-layer/DhtNodeControlLayer.cppm | 97 +++ .../PeerDescriptorStoreManager.cppm | 1 + .../createStreamPartDiscoveryLayerNode.cppm | 60 ++ .../DhtNodeDiscoveryLayer.cppm | 4 +- .../logic/ContentDeliveryLayerNode.cppm | 7 +- .../logic/ContentDeliveryRpcLocal.cppm | 8 +- .../logic/ContentDeliveryRpcRemote.cppm | 8 +- .../modules/logic/Utils.cppm | 3 +- .../logic/createContentDeliveryLayerNode.cppm | 7 +- .../logic/inspection/InspectSession.cppm | 1 + .../neighbor-discovery/HandshakeRpcLocal.cppm | 7 +- .../HandshakeRpcRemote.cppm | 5 +- .../logic/neighbor-discovery/Handshaker.cppm | 7 +- .../NeighborUpdateManager.cppm | 7 +- .../NeighborUpdateRpcLocal.cppm | 8 +- .../NeighborUpdateRpcRemote.cppm | 7 +- .../logic/propagation/FifoMapWithTTL.cppm | 6 +- .../logic/propagation/Propagation.cppm | 9 +- .../propagation/PropagationTaskStore.cppm | 5 +- .../modules/logic/proxy/ProxyClient.cppm | 12 +- .../logic/proxy/ProxyConnectionRpcLocal.cppm | 13 +- .../logic/proxy/ProxyConnectionRpcRemote.cppm | 9 +- .../TemporaryConnectionRpcLocal.cppm | 6 +- .../TemporaryConnectionRpcRemote.cppm | 7 +- .../test/unit/ContentDeliveryManagerTest.cpp | 287 ++++++++ 31 files changed, 1271 insertions(+), 40 deletions(-) create mode 100644 packages/streamr-trackerless-network/modules/ContentDeliveryManager.cppm create mode 100644 packages/streamr-trackerless-network/modules/control-layer/ControlLayerNode.cppm create mode 100644 packages/streamr-trackerless-network/modules/control-layer/DhtNodeControlLayer.cppm create mode 100644 packages/streamr-trackerless-network/modules/control-layer/createStreamPartDiscoveryLayerNode.cppm create mode 100644 packages/streamr-trackerless-network/test/unit/ContentDeliveryManagerTest.cpp diff --git a/packages/streamr-dht/modules/dht/DhtNode.cppm b/packages/streamr-dht/modules/dht/DhtNode.cppm index 69924b8c..a1326fea 100644 --- a/packages/streamr-dht/modules/dht/DhtNode.cppm +++ b/packages/streamr-dht/modules/dht/DhtNode.cppm @@ -803,6 +803,18 @@ public: co_return co_await rpcRemote.storeData(key, data); } + // TS deleteDataFromDht: a DELETE_DATA recursive operation. + folly::coro::Task deleteDataFromDht( + DhtAddress key, bool waitForCompletion) { + if (!this->abortController.getSignal().aborted) { + co_await this->recursiveOperationManager->execute( + std::move(key), + RecursiveOperation::DELETE_DATA, + std::nullopt, + waitForCompletion); + } + } + folly::coro::Task> fetchDataFromDht( const DhtAddress& key) { co_return co_await this diff --git a/packages/streamr-trackerless-network/CMakeLists.txt b/packages/streamr-trackerless-network/CMakeLists.txt index cc23530f..902149ae 100644 --- a/packages/streamr-trackerless-network/CMakeLists.txt +++ b/packages/streamr-trackerless-network/CMakeLists.txt @@ -183,6 +183,7 @@ if(NOT IOS AND STREAMR_MODULES_SUPPORTED) test/unit/NeighborUpdateRpcLocalTest.cpp test/unit/NeighborUpdateRpcRemoteTest.cpp test/unit/ContentDeliveryLayerNodeTest.cpp + test/unit/ContentDeliveryManagerTest.cpp test/unit/ContentDeliveryLayerNodeLayer1Test.cpp test/unit/PropagationScaleTest.cpp ) diff --git a/packages/streamr-trackerless-network/lint.sh b/packages/streamr-trackerless-network/lint.sh index a1ddec53..eea747bc 100755 --- a/packages/streamr-trackerless-network/lint.sh +++ b/packages/streamr-trackerless-network/lint.sh @@ -38,8 +38,12 @@ fi # ContentDeliveryLayerNodeTest.cpp (phase C3) trips the std-type # unification false positive on its own std::string locals; the # compiler builds and runs it. +# ContentDeliveryManagerTest.cpp (phase C5) trips the same std-type +# unification false positive inside the generated protobuf setter +# (set_content -> ArenaStringPtr::SetBytes); 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' | grep -v 'test/unit/ContentDeliveryLayerNodeTest.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' | grep -v 'test/unit/ContentDeliveryManagerTest.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/ContentDeliveryManager.cppm b/packages/streamr-trackerless-network/modules/ContentDeliveryManager.cppm new file mode 100644 index 00000000..6e4c0e30 --- /dev/null +++ b/packages/streamr-trackerless-network/modules/ContentDeliveryManager.cppm @@ -0,0 +1,642 @@ +// Module streamr.trackerlessnetwork.ContentDeliveryManager +// Ported from packages/trackerless-network/src/ContentDeliveryManager.ts +// (v103.8.0-rc.3): per-stream-part join/leave/broadcast orchestration — +// creates a discovery-layer node (a layer-1 DhtNode over the layer-0 +// control node), the content-delivery node, the entry-point store +// manager, split avoidance and reconnect for every joined stream part, +// and the ProxyClient for proxied stream parts. +// +// Adaptations: TS metrics (MetricsContext/RateMetric) and diagnostics +// (getDiagnosticInfo) are not ported (consistent with earlier phases). +// The TS private-client mode toggles on the ConnectionManager +// (enable/disable around proxied-only operation) have no C++ +// counterpart yet — documented follow-up. The plumtree delivery options +// are milestone-E scope. The TS fire-and-forget setImmediate() join is +// a bounded GuardedAsyncScope task. +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 +#include +#include +#include + +export module streamr.trackerlessnetwork.ContentDeliveryManager; + +import streamr.utils.CoroutineHelper; +import streamr.utils.GuardedAsyncScope; +import streamr.utils.SharedExecutors; +import streamr.trackerlessnetwork.ContentDeliveryLayerNode; +import streamr.trackerlessnetwork.protos; +import streamr.trackerlessnetwork.ControlLayerNode; +import streamr.trackerlessnetwork.createContentDeliveryLayerNode; +import streamr.trackerlessnetwork.DiscoveryLayerNode; +import streamr.trackerlessnetwork.PeerDescriptorStoreManager; +import streamr.trackerlessnetwork.ProxyClient; +import streamr.trackerlessnetwork.StreamPartNetworkSplitAvoidance; +import streamr.trackerlessnetwork.StreamPartReconnect; +import streamr.trackerlessnetwork.streamPartIdToDataKey; +import streamr.dht.ConnectionLocker; +import streamr.dht.Identifiers; +import streamr.dht.Transport; +import streamr.dht.protos; +import streamr.utils.StreamID; +import streamr.eventemitter.EventEmitter; +import streamr.logger.SLogger; +import streamr.utils.EthereumAddress; +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 ::google::protobuf::Any; +using streamr::dht::DhtAddress; +using streamr::dht::Identifiers; +using streamr::dht::connection::ConnectionLocker; +using streamr::dht::transport::Transport; +using streamr::eventemitter::Event; +using streamr::eventemitter::EventEmitter; +using streamr::logger::SLogger; +using streamr::utils::EthereumAddress; +using streamr::utils::GuardedAsyncScope; +using streamr::utils::StreamID; +using streamr::utils::StreamPartID; +using streamr::utils::StreamPartIDUtils; + +export namespace streamr::trackerlessnetwork { + +using ::dht::PeerDescriptor; +using streamr::trackerlessnetwork::controllayer::ControlLayerNode; +using streamr::trackerlessnetwork::controllayer::maxNodeCount; +using streamr::trackerlessnetwork::controllayer::PeerDescriptorStoreManager; +using streamr::trackerlessnetwork::controllayer:: + PeerDescriptorStoreManagerOptions; +using streamr::trackerlessnetwork::discoverylayer::DiscoveryLayerNode; +using streamr::trackerlessnetwork::proxy::ProxyClient; +using streamr::trackerlessnetwork::proxy::ProxyClientOptions; + +namespace contentdeliverymanagerevents { + +struct NewMessage : Event {}; + +} // namespace contentdeliverymanagerevents + +using ContentDeliveryManagerEvents = + std::tuple; + +struct ContentDeliveryManagerOptions { + std::optional streamPartitionNeighborTargetCount; + std::optional streamPartitionMinPropagationTargets; + std::optional streamPartitionMaxPropagationBufferSize; + std::optional acceptProxyConnections; + std::optional rpcRequestTimeout; + std::optional neighborUpdateInterval; + // When true, a node's OWN published messages are not delivered back + // to its local message listeners (no self-loop). Propagation to + // neighbors and duplicate detection are unaffected. Off by default. + bool suppressOwnMessageLoopback = false; + // The layer-1 discovery node factory. TS constructs the DhtNode + // inline; injected here because composing the DhtNode module graph + // in this TU exhausts clang's source locations — use + // createStreamPartDiscoveryLayerNode (its own module) as the + // implementation. + std::function( + const StreamPartID&, std::vector<::dht::PeerDescriptor>)> + createDiscoveryLayerNode; +}; + +// TS StreamPartitionInfo (types.ts), minus the deprecated field. +struct StreamPartitionInfo { + StreamPartID id; + std::vector controlLayerNeighbors; + std::vector contentDeliveryLayerNeighbors; +}; + +class ContentDeliveryManager + : public EventEmitter { +private: + // The TS StreamPartDelivery union: proxied=false carries the layer + // objects, proxied=true carries the client. + struct StreamPartDelivery { + bool proxied = false; + std::function broadcast; + std::function()> stop; + // proxied == false: + std::shared_ptr discoveryLayerNode; + std::shared_ptr node; + std::shared_ptr peerDescriptorStoreManager; + std::shared_ptr networkSplitAvoidance; + std::shared_ptr streamPartReconnect; + // proxied == true: + std::shared_ptr client; + }; + + ContentDeliveryManagerOptions options; + ControlLayerNode* controlLayerNode = nullptr; + Transport* transport = nullptr; + ConnectionLocker* connectionLocker = nullptr; + mutable std::recursive_mutex mutex; + std::map> streamParts; + std::map> + knownStreamPartEntryPoints; + bool started = false; + bool destroyed = false; + // The TS setImmediate() joins; bounded by the dht join timeout. + streamr::utils::SharedSerialExecutor joinExecutor{ + streamr::utils::SharedExecutors::worker()}; + GuardedAsyncScope joinScope; + +public: + explicit ContentDeliveryManager(ContentDeliveryManagerOptions options) + : options(std::move(options)) {} + + ~ContentDeliveryManager() override { + streamr::utils::blockingWait(this->destroy()); + } + + folly::coro::Task start( + ControlLayerNode& startedAndJoinedControlLayerNode, + Transport& transport, // NOLINT(bugprone-easily-swappable-parameters) + ConnectionLocker& connectionLocker) { + if (this->started || this->destroyed) { + co_return; + } + this->started = true; + this->controlLayerNode = &startedAndJoinedControlLayerNode; + this->transport = &transport; + this->connectionLocker = &connectionLocker; + co_return; + } + + folly::coro::Task destroy() { + { + std::scoped_lock lock(this->mutex); + if (!this->started || this->destroyed) { + co_return; + } + SLogger::trace("Destroying ContentDeliveryManager"); + this->destroyed = true; + } + this->joinScope.close(); + std::vector> parts; + { + std::scoped_lock lock(this->mutex); + for (const auto& [id, part] : this->streamParts) { + parts.push_back(part); + } + this->streamParts.clear(); + } + for (const auto& part : parts) { + co_await part->stop(); + } + this->removeAllListeners(); + this->controlLayerNode = nullptr; + this->transport = nullptr; + this->connectionLocker = nullptr; + } + + void broadcast(const StreamMessage& msg) { + const auto streamPartId = streamr::utils::toStreamPartID( + StreamID{msg.messageid().streamid()}, + static_cast(msg.messageid().streampartition())); + SLogger::debug("Broadcasting to stream part " + streamPartId); + this->joinStreamPart(streamPartId); + std::shared_ptr part; + { + std::scoped_lock lock(this->mutex); + part = this->streamParts.at(streamPartId); + } + part->broadcast(msg); + } + + folly::coro::Task leaveStreamPart(StreamPartID streamPartId) { + std::shared_ptr part; + { + std::scoped_lock lock(this->mutex); + const auto it = this->streamParts.find(streamPartId); + if (it == this->streamParts.end()) { + co_return; + } + part = it->second; + this->streamParts.erase(it); + } + co_await part->stop(); + } + + void joinStreamPart(const StreamPartID& streamPartId) { + { + std::scoped_lock lock(this->mutex); + if (this->streamParts.contains(streamPartId)) { + return; + } + } + SLogger::debug("Join stream part " + streamPartId); + std::vector knownEntryPoints; + { + std::scoped_lock lock(this->mutex); + const auto it = this->knownStreamPartEntryPoints.find(streamPartId); + if (it != this->knownStreamPartEntryPoints.end()) { + knownEntryPoints = it->second; + } + } + auto discoveryLayerNode = + this->createDiscoveryLayerNode(streamPartId, knownEntryPoints); + auto peerDescriptorStoreManager = + std::make_shared( + PeerDescriptorStoreManagerOptions{ + .key = streamPartIdToDataKey(streamPartId), + .localPeerDescriptor = this->getPeerDescriptor(), + .fetchDataFromDht = + [this](DhtAddress key) { + return this->controlLayerNode->fetchDataFromDht( + std::move(key)); + }, + .storeDataToDht = + [this](DhtAddress key, Any data) { + return this->controlLayerNode->storeDataToDht( + std::move(key), std::move(data)); + }, + .deleteDataFromDht = + [this](DhtAddress key, bool waitForCompletion) { + return this->controlLayerNode->deleteDataFromDht( + std::move(key), waitForCompletion); + }}); + auto networkSplitAvoidance = + std::make_shared( + StreamPartNetworkSplitAvoidanceOptions{ + .discoveryLayerNode = *discoveryLayerNode, + .discoverEntryPoints = [peerDescriptorStoreManager]() + -> folly::coro::Task> { + co_return co_await peerDescriptorStoreManager + ->fetchNodes(); + }}); + auto node = this->createContentDeliveryLayerNode( + streamPartId, discoveryLayerNode, [peerDescriptorStoreManager]() { + return peerDescriptorStoreManager->isLocalNodeStored(); + }); + auto streamPartReconnect = std::make_shared( + *discoveryLayerNode, *peerDescriptorStoreManager); + auto streamPart = std::make_shared(); + streamPart->proxied = false; + streamPart->discoveryLayerNode = discoveryLayerNode; + streamPart->node = node; + streamPart->peerDescriptorStoreManager = peerDescriptorStoreManager; + streamPart->networkSplitAvoidance = networkSplitAvoidance; + streamPart->streamPartReconnect = streamPartReconnect; + streamPart->broadcast = [node](const StreamMessage& msg) { + node->broadcast(msg); + }; + streamPart->stop = [streamPartReconnect, + networkSplitAvoidance, + peerDescriptorStoreManager, + node, + discoveryLayerNode]() -> folly::coro::Task { + streamPartReconnect->destroy(); + networkSplitAvoidance->destroy(); + co_await peerDescriptorStoreManager->destroy(); + node->stop(); + co_await discoveryLayerNode->stop(); + }; + { + std::scoped_lock lock(this->mutex); + this->streamParts.emplace(streamPartId, streamPart); + } + node->on( + [this](const StreamMessage& message) { + this->emit(message); + }); + discoveryLayerNode->on< + streamr::trackerlessnetwork::discoverylayer:: + discoverylayernodeevents::ManualRejoinRequired>( + [this, streamPartId, streamPartReconnect, networkSplitAvoidance]() { + if (!streamPartReconnect->isRunning() && + !networkSplitAvoidance->isRunning()) { + SLogger::debug( + "Manual rejoin required for stream part " + + streamPartId); + this->joinScope.add( + streamr::utils::co_withExecutor( + &this->joinExecutor, + streamPartReconnect->reconnect())); + } + }); + node->on( + [this, streamPartId, peerDescriptorStoreManager]() { + this->joinScope.add( + streamr::utils::co_withExecutor( + &this->joinExecutor, + this->handleEntryPointLeave( + streamPartId, peerDescriptorStoreManager))); + }); + // TS setImmediate(): detached bounded join. + this->joinScope.add( + streamr::utils::co_withExecutor( + &this->joinExecutor, + folly::coro::co_invoke( + [this, streamPartId, peerDescriptorStoreManager]() + -> folly::coro::Task { + try { + co_await this->startLayersAndJoinDht( + streamPartId, peerDescriptorStoreManager); + } catch (const std::exception& err) { + SLogger::warn( + "Failed to join to stream part " + + streamPartId + ": " + std::string(err.what())); + } + }))); + } + + folly::coro::Task setProxies( + StreamPartID streamPartId, + std::vector nodes, + ProxyDirection direction, + EthereumAddress userId, + std::optional connectionCount = std::nullopt) { + // TS TODO preserved: explicit default value for + // "acceptProxyConnections" or make it required. + if (this->options.acceptProxyConnections.value_or(false)) { + throw std::runtime_error( + "cannot set proxies when acceptProxyConnections=true"); + } + const bool enable = !nodes.empty() && + (!connectionCount.has_value() || connectionCount.value() > 0); + if (enable) { + std::shared_ptr client; + bool alreadyProxied = false; + { + std::scoped_lock lock(this->mutex); + alreadyProxied = this->isProxiedStreamPart(streamPartId); + if (alreadyProxied) { + client = this->streamParts.at(streamPartId)->client; + } + } + if (!alreadyProxied) { + client = this->createProxyClient(streamPartId); + auto streamPart = std::make_shared(); + streamPart->proxied = true; + streamPart->client = client; + streamPart->broadcast = [client](const StreamMessage& msg) { + client->broadcast(msg); + }; + streamPart->stop = [client]() -> folly::coro::Task { + client->stop(); + co_return; + }; + { + std::scoped_lock lock(this->mutex); + this->streamParts.emplace(streamPartId, streamPart); + } + client->on( + [this](const StreamMessage& message) { + this->emit( + message); + }); + // TS enables ConnectionManager private-client mode when + // every stream part is proxied; no C++ counterpart yet. + client->start(); + } + // ProxyClient::setProxies is synchronous (the shared-library + // API); TS awaits and discards the result. Connection errors + // are reported through the returned pair, which only the + // shared library consumes. + client->setProxies(nodes, direction, userId, connectionCount); + } else { + co_await this->leaveStreamPart(streamPartId); + } + } + + folly::coro::Task inspect( + PeerDescriptor peerDescriptor, StreamPartID streamPartId) { + std::shared_ptr part; + { + std::scoped_lock lock(this->mutex); + const auto it = this->streamParts.find(streamPartId); + if (it != this->streamParts.end() && !it->second->proxied) { + part = it->second; + } + } + if (part) { + co_return co_await part->node->inspect(std::move(peerDescriptor)); + } + co_return false; + } + + [[nodiscard]] std::vector getNodeInfo() const { + std::scoped_lock lock(this->mutex); + std::vector infos; + for (const auto& [streamPartId, part] : this->streamParts) { + if (part->proxied) { + continue; + } + infos.push_back( + StreamPartitionInfo{ + .id = streamPartId, + .controlLayerNeighbors = + part->discoveryLayerNode->getNeighbors(), + .contentDeliveryLayerNeighbors = part->node->getInfos()}); + } + return infos; + } + + void setStreamPartEntryPoints( + const StreamPartID& streamPartId, + std::vector entryPoints) { + std::scoped_lock lock(this->mutex); + this->knownStreamPartEntryPoints[streamPartId] = std::move(entryPoints); + } + + [[nodiscard]] bool isProxiedStreamPart( + const StreamPartID& streamPartId, + std::optional direction = std::nullopt) const { + std::scoped_lock lock(this->mutex); + const auto it = this->streamParts.find(streamPartId); + return it != this->streamParts.end() && it->second->proxied && + (!direction.has_value() || + it->second->client->getDirection() == direction.value()); + } + + [[nodiscard]] bool hasStreamPart(const StreamPartID& streamPartId) const { + std::scoped_lock lock(this->mutex); + return this->streamParts.contains(streamPartId); + } + + [[nodiscard]] PeerDescriptor getPeerDescriptor() const { + return this->controlLayerNode->getLocalPeerDescriptor(); + } + + [[nodiscard]] DhtAddress getNodeId() const { + return Identifiers::getNodeIdFromPeerDescriptor( + this->getPeerDescriptor()); + } + + [[nodiscard]] std::vector getNeighbors( + const StreamPartID& streamPartId) const { + std::shared_ptr part; + { + std::scoped_lock lock(this->mutex); + const auto it = this->streamParts.find(streamPartId); + if (it == this->streamParts.end() || it->second->proxied) { + return {}; + } + part = it->second; + } + std::vector ids; + for (const auto& descriptor : part->node->getNeighbors()) { + ids.push_back(Identifiers::getNodeIdFromPeerDescriptor(descriptor)); + } + return ids; + } + + [[nodiscard]] std::vector getStreamParts() const { + std::scoped_lock lock(this->mutex); + std::vector ids; + for (const auto& [id, part] : this->streamParts) { + ids.push_back(id); + } + return ids; + } + +private: + folly::coro::Task handleEntryPointLeave( + StreamPartID streamPartId, + std::shared_ptr + peerDescriptorStoreManager) { + { + std::scoped_lock lock(this->mutex); + if (this->destroyed || + peerDescriptorStoreManager->isLocalNodeStored() || + this->knownStreamPartEntryPoints.contains(streamPartId)) { + co_return; + } + } + const auto entryPoints = + co_await peerDescriptorStoreManager->fetchNodes(); + if (entryPoints.size() < maxNodeCount) { + co_await peerDescriptorStoreManager->storeAndKeepLocalNode(); + } + } + + folly::coro::Task startLayersAndJoinDht( + StreamPartID streamPartId, + std::shared_ptr + peerDescriptorStoreManager) { + SLogger::debug( + "Start layers and join DHT for stream part " + streamPartId); + std::shared_ptr streamPart; + std::optional> knownEntryPoints; + { + std::scoped_lock lock(this->mutex); + const auto it = this->streamParts.find(streamPartId); + if (it == this->streamParts.end() || it->second->proxied) { + // leaveStreamPart has been called (or setProxies since). + co_return; + } + streamPart = it->second; + const auto epIt = + this->knownStreamPartEntryPoints.find(streamPartId); + if (epIt != this->knownStreamPartEntryPoints.end()) { + knownEntryPoints = epIt->second; + } + } + co_await streamPart->discoveryLayerNode->start(); + co_await streamPart->node->start(); + if (knownEntryPoints.has_value()) { + co_await folly::coro::collectAll( + streamPart->discoveryLayerNode->joinDht( + knownEntryPoints.value()), + streamPart->discoveryLayerNode->joinRing()); + } else { + auto entryPoints = + co_await peerDescriptorStoreManager->fetchNodes(); + co_await folly::coro::collectAll( + streamPart->discoveryLayerNode->joinDht( + sample(entryPoints, minNeighborCount)), + streamPart->discoveryLayerNode->joinRing()); + if (entryPoints.size() < maxNodeCount) { + co_await peerDescriptorStoreManager->storeAndKeepLocalNode(); + if (streamPart->discoveryLayerNode->getNeighborCount() < + minNeighborCount) { + auto networkSplitAvoidance = + streamPart->networkSplitAvoidance; + this->joinScope.add( + streamr::utils::co_withExecutor( + &this->joinExecutor, + folly::coro::co_invoke( + [networkSplitAvoidance]() + -> folly::coro::Task { + co_await networkSplitAvoidance + ->avoidNetworkSplit(); + }))); + } + } + } + } + + // lodash sampleSize: up to n random elements. + static std::vector sample( + std::vector input, size_t count) { + static thread_local std::mt19937 generator{std::random_device{}()}; + std::shuffle(input.begin(), input.end(), generator); + if (input.size() > count) { + input.resize(count); + } + return input; + } + + std::shared_ptr createDiscoveryLayerNode( + const StreamPartID& streamPartId, + std::vector entryPoints) { + return this->options.createDiscoveryLayerNode( + streamPartId, std::move(entryPoints)); + } + + std::shared_ptr createContentDeliveryLayerNode( + const StreamPartID& streamPartId, + std::shared_ptr discoveryLayerNode, + std::function isLocalNodeEntryPoint) { + return streamr::trackerlessnetwork::createContentDeliveryLayerNode( + ContentDeliveryLayerNodeOptions{ + .streamPartId = streamPartId, + .discoveryLayerNode = std::move(discoveryLayerNode), + .transport = this->transport, + .connectionLocker = this->connectionLocker, + .localPeerDescriptor = + this->controlLayerNode->getLocalPeerDescriptor(), + .isLocalNodeEntryPoint = std::move(isLocalNodeEntryPoint), + .neighborTargetCount = + this->options.streamPartitionNeighborTargetCount, + .minPropagationTargets = + this->options.streamPartitionMinPropagationTargets, + .maxPropagationBufferSize = + this->options.streamPartitionMaxPropagationBufferSize, + .acceptProxyConnections = this->options.acceptProxyConnections, + .neighborUpdateInterval = this->options.neighborUpdateInterval, + .rpcRequestTimeout = this->options.rpcRequestTimeout, + .suppressOwnMessageLoopback = + this->options.suppressOwnMessageLoopback}); + } + + std::shared_ptr createProxyClient( + const StreamPartID& streamPartId) { + return std::make_shared(ProxyClientOptions{ + .transport = *this->transport, + .localPeerDescriptor = + this->controlLayerNode->getLocalPeerDescriptor(), + .streamPartId = streamPartId, + .connectionLocker = *this->connectionLocker, + .minPropagationTargets = + this->options.streamPartitionMinPropagationTargets}); + } +}; + +} // namespace streamr::trackerlessnetwork diff --git a/packages/streamr-trackerless-network/modules/StreamPartNetworkSplitAvoidance.cppm b/packages/streamr-trackerless-network/modules/StreamPartNetworkSplitAvoidance.cppm index 37f87e16..8b1dc4a7 100644 --- a/packages/streamr-trackerless-network/modules/StreamPartNetworkSplitAvoidance.cppm +++ b/packages/streamr-trackerless-network/modules/StreamPartNetworkSplitAvoidance.cppm @@ -28,6 +28,7 @@ module; #include #include // IWYU pragma: keep +#include export module streamr.trackerlessnetwork.StreamPartNetworkSplitAvoidance; diff --git a/packages/streamr-trackerless-network/modules/control-layer/ControlLayerNode.cppm b/packages/streamr-trackerless-network/modules/control-layer/ControlLayerNode.cppm new file mode 100644 index 00000000..8e38c46e --- /dev/null +++ b/packages/streamr-trackerless-network/modules/control-layer/ControlLayerNode.cppm @@ -0,0 +1,58 @@ +// Module streamr.trackerlessnetwork.ControlLayerNode +// Ported from packages/trackerless-network/src/control-layer/ +// ControlLayerNode.ts (v103.8.0-rc.3): the layer-0 facade the +// ContentDeliveryManager and NetworkStack talk to. TS extends +// ITransport structurally; here asTransport() exposes the transport +// identity nominally (the layer-1 discovery nodes use the layer-0 node +// itself as their signalling transport). +module; + +// Coroutine definitions need std::coroutine_traits declared in THIS +// translation unit; it cannot arrive through an imported BMI. +#include // IWYU pragma: keep + +#include + +export module streamr.trackerlessnetwork.ControlLayerNode; + +import streamr.utils.CoroutineHelper; +import streamr.dht.ConnectionsView; +import streamr.dht.Identifiers; +import streamr.dht.Transport; +import streamr.dht.protos; + +using streamr::dht::DhtAddress; +using streamr::dht::connection::ConnectionsView; +using streamr::dht::transport::Transport; + +export namespace streamr::trackerlessnetwork::controllayer { + +using ::dht::DataEntry; +using ::dht::PeerDescriptor; +using ::google::protobuf::Any; + +class ControlLayerNode { +public: + virtual ~ControlLayerNode() = default; + + virtual folly::coro::Task joinDht( + std::vector entryPointDescriptors) = 0; + [[nodiscard]] virtual bool hasJoined() = 0; + [[nodiscard]] virtual PeerDescriptor getLocalPeerDescriptor() = 0; + virtual folly::coro::Task> fetchDataFromDht( + DhtAddress key) = 0; + virtual folly::coro::Task> storeDataToDht( + DhtAddress key, Any data) = 0; + virtual folly::coro::Task deleteDataFromDht( + DhtAddress key, bool waitForCompletion) = 0; + virtual folly::coro::Task waitForNetworkConnectivity() = 0; + // The layer-0 node as a Transport (TS passes the node itself where a + // transport is expected). + [[nodiscard]] virtual Transport* asTransport() = 0; + [[nodiscard]] virtual std::vector getNeighbors() = 0; + [[nodiscard]] virtual ConnectionsView* getConnectionsView() = 0; + virtual folly::coro::Task start() = 0; + virtual folly::coro::Task stop() = 0; +}; + +} // namespace streamr::trackerlessnetwork::controllayer diff --git a/packages/streamr-trackerless-network/modules/control-layer/DhtNodeControlLayer.cppm b/packages/streamr-trackerless-network/modules/control-layer/DhtNodeControlLayer.cppm new file mode 100644 index 00000000..1f5289ae --- /dev/null +++ b/packages/streamr-trackerless-network/modules/control-layer/DhtNodeControlLayer.cppm @@ -0,0 +1,97 @@ +// Module streamr.trackerlessnetwork.DhtNodeControlLayer +// The TS code passes a layer-0 DhtNode wherever a ControlLayerNode is +// expected (structural typing); this adapter provides that shape +// nominally over streamr::dht::DhtNode, mirroring DhtNodeDiscoveryLayer. +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 + +export module streamr.trackerlessnetwork.DhtNodeControlLayer; + +import streamr.utils.CoroutineHelper; +import streamr.trackerlessnetwork.ControlLayerNode; +import streamr.dht.ConnectionsView; +import streamr.dht.DhtNode; +import streamr.dht.Identifiers; +import streamr.dht.Transport; +import streamr.dht.protos; + +using streamr::dht::DhtAddress; +using streamr::dht::DhtNode; +using streamr::dht::connection::ConnectionsView; +using streamr::dht::transport::Transport; + +export namespace streamr::trackerlessnetwork::controllayer { + +class DhtNodeControlLayer : public ControlLayerNode { +private: + std::shared_ptr dhtNode; + +public: + explicit DhtNodeControlLayer(std::shared_ptr dhtNode) + : dhtNode(std::move(dhtNode)) {} + + [[nodiscard]] DhtNode& getDhtNode() { return *this->dhtNode; } + + folly::coro::Task joinDht( + std::vector entryPointDescriptors) override { + co_await this->dhtNode->joinDht(std::move(entryPointDescriptors)); + } + + [[nodiscard]] bool hasJoined() override { + return this->dhtNode->hasJoined(); + } + + [[nodiscard]] PeerDescriptor getLocalPeerDescriptor() override { + return this->dhtNode->getLocalPeerDescriptor(); + } + + folly::coro::Task> fetchDataFromDht( + DhtAddress key) override { + co_return co_await this->dhtNode->fetchDataFromDht(key); + } + + folly::coro::Task> storeDataToDht( + DhtAddress key, Any data) override { + co_return co_await this->dhtNode->storeDataToDht( + std::move(key), std::move(data)); + } + + folly::coro::Task deleteDataFromDht( + DhtAddress key, bool waitForCompletion) override { + co_await this->dhtNode->deleteDataFromDht( + std::move(key), waitForCompletion); + } + + folly::coro::Task waitForNetworkConnectivity() override { + co_await this->dhtNode->waitForNetworkConnectivity(); + } + + [[nodiscard]] Transport* asTransport() override { + return this->dhtNode.get(); + } + + [[nodiscard]] std::vector getNeighbors() override { + return this->dhtNode->getNeighbors(); + } + + [[nodiscard]] ConnectionsView* getConnectionsView() override { + return this->dhtNode->getConnectionsView(); + } + + folly::coro::Task start() override { + co_await this->dhtNode->start(); + } + + folly::coro::Task stop() override { + this->dhtNode->stop(); + co_return; + } +}; + +} // namespace streamr::trackerlessnetwork::controllayer diff --git a/packages/streamr-trackerless-network/modules/control-layer/PeerDescriptorStoreManager.cppm b/packages/streamr-trackerless-network/modules/control-layer/PeerDescriptorStoreManager.cppm index 6761c6e1..e42ef62f 100644 --- a/packages/streamr-trackerless-network/modules/control-layer/PeerDescriptorStoreManager.cppm +++ b/packages/streamr-trackerless-network/modules/control-layer/PeerDescriptorStoreManager.cppm @@ -22,6 +22,7 @@ module; #include // IWYU pragma: keep +#include #include export module streamr.trackerlessnetwork.PeerDescriptorStoreManager; diff --git a/packages/streamr-trackerless-network/modules/control-layer/createStreamPartDiscoveryLayerNode.cppm b/packages/streamr-trackerless-network/modules/control-layer/createStreamPartDiscoveryLayerNode.cppm new file mode 100644 index 00000000..cf3163dd --- /dev/null +++ b/packages/streamr-trackerless-network/modules/control-layer/createStreamPartDiscoveryLayerNode.cppm @@ -0,0 +1,60 @@ +// Module streamr.trackerlessnetwork.createStreamPartDiscoveryLayerNode +// The layer-1 discovery node construction from TS +// ContentDeliveryManager.createDiscoveryLayerNode (v103.8.0-rc.3), +// extracted into its own module: composing the DhtNode module graph +// inside ContentDeliveryManager.cppm exhausts clang's per-TU +// source-location space, so the manager takes this as an injected +// factory instead (NetworkStack and the tests supply it). +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 + +export module streamr.trackerlessnetwork.createStreamPartDiscoveryLayerNode; + +import streamr.utils.CoroutineHelper; +import streamr.trackerlessnetwork.ControlLayerNode; +import streamr.trackerlessnetwork.DhtNodeDiscoveryLayer; +import streamr.trackerlessnetwork.DiscoveryLayerNode; +import streamr.dht.DhtNode; +import streamr.dht.Identifiers; +import streamr.dht.protos; +import streamr.utils.StreamPartID; + +using streamr::dht::DhtNode; +using streamr::dht::DhtNodeOptions; +using streamr::dht::ServiceID; +using streamr::utils::StreamPartID; + +export namespace streamr::trackerlessnetwork::controllayer { + +using ::dht::PeerDescriptor; +using streamr::trackerlessnetwork::discoverylayer::DhtNodeDiscoveryLayer; +using streamr::trackerlessnetwork::discoverylayer::DiscoveryLayerNode; + +inline std::shared_ptr createStreamPartDiscoveryLayerNode( + const StreamPartID& streamPartId, + std::vector entryPoints, + ControlLayerNode& controlLayerNode) { + auto dhtNode = std::make_shared(DhtNodeOptions{ + .serviceId = ServiceID{"layer1::" + streamPartId}, + // TS TODO preserved: use options options or named constants? + .numberOfNodesPerKBucket = 4, + .dhtJoinTimeout = std::chrono::milliseconds(20000), + .neighborPingLimit = 16, + // TS EXISTING_CONNECTION_TIMEOUT (RpcRemote). + .rpcRequestTimeout = std::chrono::milliseconds(5000), + .transport = controlLayerNode.asTransport(), + .connectionsView = controlLayerNode.getConnectionsView(), + .peerDescriptor = controlLayerNode.getLocalPeerDescriptor(), + .entryPoints = std::move(entryPoints)}); + return std::make_shared(std::move(dhtNode)); +} + +} // namespace streamr::trackerlessnetwork::controllayer diff --git a/packages/streamr-trackerless-network/modules/discovery-layer/DhtNodeDiscoveryLayer.cppm b/packages/streamr-trackerless-network/modules/discovery-layer/DhtNodeDiscoveryLayer.cppm index 49dfeaa5..b15eb979 100644 --- a/packages/streamr-trackerless-network/modules/discovery-layer/DhtNodeDiscoveryLayer.cppm +++ b/packages/streamr-trackerless-network/modules/discovery-layer/DhtNodeDiscoveryLayer.cppm @@ -10,10 +10,12 @@ module; // translation unit; it cannot arrive through an imported BMI. #include // IWYU pragma: keep +#include #include #include +#include +#include #include -#include "packages/dht/protos/DhtRpc.pb.h" export module streamr.trackerlessnetwork.DhtNodeDiscoveryLayer; diff --git a/packages/streamr-trackerless-network/modules/logic/ContentDeliveryLayerNode.cppm b/packages/streamr-trackerless-network/modules/logic/ContentDeliveryLayerNode.cppm index aa49150d..5553405e 100644 --- a/packages/streamr-trackerless-network/modules/logic/ContentDeliveryLayerNode.cppm +++ b/packages/streamr-trackerless-network/modules/logic/ContentDeliveryLayerNode.cppm @@ -21,6 +21,7 @@ module; #include #include +#include #include #include #include @@ -28,12 +29,14 @@ module; #include #include #include +#include +#include #include -#include "packages/dht/protos/DhtRpc.pb.h" -#include "packages/network/protos/NetworkRpc.pb.h" export module streamr.trackerlessnetwork.ContentDeliveryLayerNode; +import streamr.trackerlessnetwork.protos; + import streamr.utils.CoroutineHelper; import streamr.trackerlessnetwork.ContentDeliveryRpcLocal; import streamr.trackerlessnetwork.ContentDeliveryRpcRemote; diff --git a/packages/streamr-trackerless-network/modules/logic/ContentDeliveryRpcLocal.cppm b/packages/streamr-trackerless-network/modules/logic/ContentDeliveryRpcLocal.cppm index ae6e7335..f12acfaa 100644 --- a/packages/streamr-trackerless-network/modules/logic/ContentDeliveryRpcLocal.cppm +++ b/packages/streamr-trackerless-network/modules/logic/ContentDeliveryRpcLocal.cppm @@ -3,13 +3,15 @@ // (MODERNIZATION.md Phase 2.6): this file is now the source of truth. module; +#include #include - -#include "packages/dht/protos/DhtRpc.pb.h" -#include "packages/network/protos/NetworkRpc.pb.h" +#include +#include export module streamr.trackerlessnetwork.ContentDeliveryRpcLocal; +import streamr.trackerlessnetwork.protos; + import streamr.trackerlessnetwork.NetworkRpcServer; import streamr.dht.DhtCallContext; import streamr.dht.Identifiers; diff --git a/packages/streamr-trackerless-network/modules/logic/ContentDeliveryRpcRemote.cppm b/packages/streamr-trackerless-network/modules/logic/ContentDeliveryRpcRemote.cppm index 2b545248..bd0c0d0e 100644 --- a/packages/streamr-trackerless-network/modules/logic/ContentDeliveryRpcRemote.cppm +++ b/packages/streamr-trackerless-network/modules/logic/ContentDeliveryRpcRemote.cppm @@ -7,12 +7,16 @@ module; // translation unit; it cannot arrive through an imported BMI. #include // IWYU pragma: keep +#include +#include +#include #include -#include "packages/dht/protos/DhtRpc.pb.h" -#include "packages/network/protos/NetworkRpc.pb.h" +#include export module streamr.trackerlessnetwork.ContentDeliveryRpcRemote; +import streamr.trackerlessnetwork.protos; + import streamr.utils.CoroutineHelper; import streamr.trackerlessnetwork.NetworkRpcClient; import streamr.dht.DhtCallContext; diff --git a/packages/streamr-trackerless-network/modules/logic/Utils.cppm b/packages/streamr-trackerless-network/modules/logic/Utils.cppm index 77d56fea..01e39f66 100644 --- a/packages/streamr-trackerless-network/modules/logic/Utils.cppm +++ b/packages/streamr-trackerless-network/modules/logic/Utils.cppm @@ -7,10 +7,11 @@ module; #include #include #include -#include "packages/network/protos/NetworkRpc.pb.h" export module streamr.trackerlessnetwork.Utils; +import streamr.trackerlessnetwork.protos; + import streamr.dht.Identifiers; import streamr.trackerlessnetwork.DuplicateMessageDetector; diff --git a/packages/streamr-trackerless-network/modules/logic/createContentDeliveryLayerNode.cppm b/packages/streamr-trackerless-network/modules/logic/createContentDeliveryLayerNode.cppm index e51613ec..bd1bfad6 100644 --- a/packages/streamr-trackerless-network/modules/logic/createContentDeliveryLayerNode.cppm +++ b/packages/streamr-trackerless-network/modules/logic/createContentDeliveryLayerNode.cppm @@ -13,17 +13,20 @@ 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.createContentDeliveryLayerNode; +import streamr.trackerlessnetwork.protos; + import streamr.utils.CoroutineHelper; import streamr.trackerlessnetwork.ContentDeliveryLayerNode; import streamr.trackerlessnetwork.DiscoveryLayerNode; diff --git a/packages/streamr-trackerless-network/modules/logic/inspection/InspectSession.cppm b/packages/streamr-trackerless-network/modules/logic/inspection/InspectSession.cppm index dfda1c68..b2846df1 100644 --- a/packages/streamr-trackerless-network/modules/logic/inspection/InspectSession.cppm +++ b/packages/streamr-trackerless-network/modules/logic/inspection/InspectSession.cppm @@ -11,6 +11,7 @@ module; #include #include #include +#include #include #include #include diff --git a/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/HandshakeRpcLocal.cppm b/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/HandshakeRpcLocal.cppm index f937d119..c9d4b75d 100644 --- a/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/HandshakeRpcLocal.cppm +++ b/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/HandshakeRpcLocal.cppm @@ -16,16 +16,19 @@ module; // translation unit; it cannot arrive through an imported BMI. #include // IWYU pragma: keep +#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.HandshakeRpcLocal; +import streamr.trackerlessnetwork.protos; + import streamr.utils.CoroutineHelper; import streamr.utils.GuardedAsyncScope; import streamr.utils.SharedExecutors; diff --git a/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/HandshakeRpcRemote.cppm b/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/HandshakeRpcRemote.cppm index 2da5f2b1..6beba17d 100644 --- a/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/HandshakeRpcRemote.cppm +++ b/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/HandshakeRpcRemote.cppm @@ -12,12 +12,13 @@ module; #include #include #include +#include #include -#include "packages/dht/protos/DhtRpc.pb.h" -#include "packages/network/protos/NetworkRpc.pb.h" export module streamr.trackerlessnetwork.HandshakeRpcRemote; +import streamr.trackerlessnetwork.protos; + import streamr.utils.CoroutineHelper; import streamr.trackerlessnetwork.NetworkRpcClient; import streamr.dht.DhtCallContext; diff --git a/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/Handshaker.cppm b/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/Handshaker.cppm index 8c95f2dc..fea4ce52 100644 --- a/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/Handshaker.cppm +++ b/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/Handshaker.cppm @@ -12,16 +12,19 @@ module; #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.Handshaker; +import streamr.trackerlessnetwork.protos; + import streamr.utils.CoroutineHelper; import streamr.trackerlessnetwork.ContentDeliveryRpcRemote; import streamr.trackerlessnetwork.HandshakeRpcLocal; diff --git a/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/NeighborUpdateManager.cppm b/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/NeighborUpdateManager.cppm index d34d3aef..31e2685a 100644 --- a/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/NeighborUpdateManager.cppm +++ b/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/NeighborUpdateManager.cppm @@ -14,14 +14,17 @@ 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.NeighborUpdateManager; +import streamr.trackerlessnetwork.protos; + import streamr.utils.CoroutineHelper; import streamr.utils.AbortController; import streamr.utils.GuardedAsyncScope; diff --git a/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/NeighborUpdateRpcLocal.cppm b/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/NeighborUpdateRpcLocal.cppm index 44803de4..a24a7802 100644 --- a/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/NeighborUpdateRpcLocal.cppm +++ b/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/NeighborUpdateRpcLocal.cppm @@ -7,14 +7,18 @@ module; #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.NeighborUpdateRpcLocal; +import streamr.trackerlessnetwork.protos; + import streamr.trackerlessnetwork.ContentDeliveryRpcRemote; import streamr.trackerlessnetwork.NeighborFinder; import streamr.trackerlessnetwork.NetworkRpcClient; diff --git a/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/NeighborUpdateRpcRemote.cppm b/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/NeighborUpdateRpcRemote.cppm index 8af4f708..a6f420b9 100644 --- a/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/NeighborUpdateRpcRemote.cppm +++ b/packages/streamr-trackerless-network/modules/logic/neighbor-discovery/NeighborUpdateRpcRemote.cppm @@ -9,13 +9,16 @@ module; // translation unit; it cannot arrive through an imported BMI. #include // IWYU pragma: keep +#include +#include #include +#include #include -#include "packages/dht/protos/DhtRpc.pb.h" -#include "packages/network/protos/NetworkRpc.pb.h" export module streamr.trackerlessnetwork.NeighborUpdateRpcRemote; +import streamr.trackerlessnetwork.protos; + import streamr.utils.CoroutineHelper; import streamr.trackerlessnetwork.NetworkRpcClient; import streamr.dht.DhtCallContext; diff --git a/packages/streamr-trackerless-network/modules/logic/propagation/FifoMapWithTTL.cppm b/packages/streamr-trackerless-network/modules/logic/propagation/FifoMapWithTTL.cppm index 359adf6a..13fd95f3 100644 --- a/packages/streamr-trackerless-network/modules/logic/propagation/FifoMapWithTTL.cppm +++ b/packages/streamr-trackerless-network/modules/logic/propagation/FifoMapWithTTL.cppm @@ -4,12 +4,16 @@ module; #include +#include #include #include #include #include #include -#include "packages/network/protos/NetworkRpc.pb.h" +#include +#include +#include +#include export module streamr.trackerlessnetwork.FifoMapWithTTL; diff --git a/packages/streamr-trackerless-network/modules/logic/propagation/Propagation.cppm b/packages/streamr-trackerless-network/modules/logic/propagation/Propagation.cppm index 2f070ec1..d7d9aee7 100644 --- a/packages/streamr-trackerless-network/modules/logic/propagation/Propagation.cppm +++ b/packages/streamr-trackerless-network/modules/logic/propagation/Propagation.cppm @@ -13,16 +13,21 @@ 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.Propagation; +import streamr.trackerlessnetwork.protos; + import streamr.utils.CoroutineHelper; import streamr.utils.GuardedAsyncScope; import streamr.utils.SharedExecutors; diff --git a/packages/streamr-trackerless-network/modules/logic/propagation/PropagationTaskStore.cppm b/packages/streamr-trackerless-network/modules/logic/propagation/PropagationTaskStore.cppm index f8c438ee..f56b672d 100644 --- a/packages/streamr-trackerless-network/modules/logic/propagation/PropagationTaskStore.cppm +++ b/packages/streamr-trackerless-network/modules/logic/propagation/PropagationTaskStore.cppm @@ -5,13 +5,16 @@ module; #include +#include #include #include #include -#include "packages/network/protos/NetworkRpc.pb.h" +#include export module streamr.trackerlessnetwork.PropagationTaskStore; +import streamr.trackerlessnetwork.protos; + import streamr.dht.Identifiers; import streamr.trackerlessnetwork.FifoMapWithTTL; diff --git a/packages/streamr-trackerless-network/modules/logic/proxy/ProxyClient.cppm b/packages/streamr-trackerless-network/modules/logic/proxy/ProxyClient.cppm index 16e68428..840c7b5e 100644 --- a/packages/streamr-trackerless-network/modules/logic/proxy/ProxyClient.cppm +++ b/packages/streamr-trackerless-network/modules/logic/proxy/ProxyClient.cppm @@ -8,20 +8,28 @@ module; // an imported BMI. #include // IWYU pragma: keep +#include +#include #include #include +#include #include #include +#include +#include #include +#include +#include +#include // Textual: entities reached only through an imported module's global // module fragment are not reliably reachable; this unit's code calls // streamr::utils::blockingWait and std::mt19937 directly. (The former // header received both transitively from the headers it included.) -#include "packages/dht/protos/DhtRpc.pb.h" -#include "packages/network/protos/NetworkRpc.pb.h" export module streamr.trackerlessnetwork.ProxyClient; +import streamr.trackerlessnetwork.protos; + import streamr.utils.CoroutineHelper; import streamr.dht.DhtCallContext; import streamr.dht.ListeningRpcCommunicator; diff --git a/packages/streamr-trackerless-network/modules/logic/proxy/ProxyConnectionRpcLocal.cppm b/packages/streamr-trackerless-network/modules/logic/proxy/ProxyConnectionRpcLocal.cppm index 3eb2bca3..78ce2a9b 100644 --- a/packages/streamr-trackerless-network/modules/logic/proxy/ProxyConnectionRpcLocal.cppm +++ b/packages/streamr-trackerless-network/modules/logic/proxy/ProxyConnectionRpcLocal.cppm @@ -7,12 +7,19 @@ module; // that defines OR instantiates a coroutine; it cannot arrive through // an imported BMI. #include // IWYU pragma: keep - -#include "packages/dht/protos/DhtRpc.pb.h" -#include "packages/network/protos/NetworkRpc.pb.h" +#include +#include +#include +#include +#include +#include +#include +#include export module streamr.trackerlessnetwork.ProxyConnectionRpcLocal; +import streamr.trackerlessnetwork.protos; + import streamr.utils.CoroutineHelper; import streamr.trackerlessnetwork.NetworkRpcServer; import streamr.logger.SLogger; diff --git a/packages/streamr-trackerless-network/modules/logic/proxy/ProxyConnectionRpcRemote.cppm b/packages/streamr-trackerless-network/modules/logic/proxy/ProxyConnectionRpcRemote.cppm index b8529233..6d1f4d07 100644 --- a/packages/streamr-trackerless-network/modules/logic/proxy/ProxyConnectionRpcRemote.cppm +++ b/packages/streamr-trackerless-network/modules/logic/proxy/ProxyConnectionRpcRemote.cppm @@ -5,13 +5,16 @@ module; // Coroutine definitions need std::coroutine_traits declared in THIS // translation unit; it cannot arrive through an imported BMI. +#include #include // IWYU pragma: keep - -#include "packages/dht/protos/DhtRpc.pb.h" -#include "packages/network/protos/NetworkRpc.pb.h" +#include +#include +#include export module streamr.trackerlessnetwork.ProxyConnectionRpcRemote; +import streamr.trackerlessnetwork.protos; + import streamr.utils.CoroutineHelper; import streamr.trackerlessnetwork.NetworkRpcClient; import streamr.dht.DhtCallContext; diff --git a/packages/streamr-trackerless-network/modules/logic/temporary-connection/TemporaryConnectionRpcLocal.cppm b/packages/streamr-trackerless-network/modules/logic/temporary-connection/TemporaryConnectionRpcLocal.cppm index a3850476..dbb672af 100644 --- a/packages/streamr-trackerless-network/modules/logic/temporary-connection/TemporaryConnectionRpcLocal.cppm +++ b/packages/streamr-trackerless-network/modules/logic/temporary-connection/TemporaryConnectionRpcLocal.cppm @@ -6,13 +6,15 @@ // and weak-locking the underlying connection while it is in use. module; +#include #include #include -#include "packages/dht/protos/DhtRpc.pb.h" -#include "packages/network/protos/NetworkRpc.pb.h" +#include export module streamr.trackerlessnetwork.TemporaryConnectionRpcLocal; +import streamr.trackerlessnetwork.protos; + import streamr.trackerlessnetwork.NetworkRpcServer; import streamr.trackerlessnetwork.NetworkRpcClient; import streamr.trackerlessnetwork.ContentDeliveryRpcRemote; diff --git a/packages/streamr-trackerless-network/modules/logic/temporary-connection/TemporaryConnectionRpcRemote.cppm b/packages/streamr-trackerless-network/modules/logic/temporary-connection/TemporaryConnectionRpcRemote.cppm index 63cf3506..81ac792d 100644 --- a/packages/streamr-trackerless-network/modules/logic/temporary-connection/TemporaryConnectionRpcRemote.cppm +++ b/packages/streamr-trackerless-network/modules/logic/temporary-connection/TemporaryConnectionRpcRemote.cppm @@ -10,12 +10,15 @@ module; // translation unit; it cannot arrive through an imported BMI. #include // IWYU pragma: keep +#include +#include #include -#include "packages/dht/protos/DhtRpc.pb.h" -#include "packages/network/protos/NetworkRpc.pb.h" +#include export module streamr.trackerlessnetwork.TemporaryConnectionRpcRemote; +import streamr.trackerlessnetwork.protos; + import streamr.utils.CoroutineHelper; import streamr.trackerlessnetwork.NetworkRpcClient; import streamr.dht.DhtCallContext; diff --git a/packages/streamr-trackerless-network/test/unit/ContentDeliveryManagerTest.cpp b/packages/streamr-trackerless-network/test/unit/ContentDeliveryManagerTest.cpp new file mode 100644 index 00000000..ed66a94b --- /dev/null +++ b/packages/streamr-trackerless-network/test/unit/ContentDeliveryManagerTest.cpp @@ -0,0 +1,287 @@ +// Ported from packages/trackerless-network/test/integration/ +// ContentDeliveryManager.test.ts (v103.8.0-rc.3): two managers over +// layer-0 DhtNodes on a simulator — joining, pub/sub, multi-stream, +// leaving, and RTT collection. +// +// NB: TestUtils and the textual pb.h are avoided — this TU composes the +// full DhtNode + manager module graph and additional BMIs exhaust +// clang's per-TU source-location space (see the C3 test files). +#include +#include +#include +#include +#include +#include + +#include // IWYU pragma: keep + +import streamr.utils.CoroutineHelper; +import streamr.trackerlessnetwork.ContentDeliveryManager; +import streamr.trackerlessnetwork.DhtNodeControlLayer; +import streamr.trackerlessnetwork.createStreamPartDiscoveryLayerNode; +import streamr.trackerlessnetwork.DiscoveryLayerNode; +import streamr.trackerlessnetwork.protos; +import streamr.dht.DhtNode; +import streamr.dht.Identifiers; +import streamr.dht.Simulator; +import streamr.dht.SimulatorTransport; +import streamr.dht.protos; +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::connection::simulator::LatencyType; +using streamr::dht::connection::simulator::Simulator; +using streamr::dht::connection::simulator::SimulatorTransport; +using streamr::trackerlessnetwork::ContentDeliveryManager; +using streamr::trackerlessnetwork::ContentDeliveryManagerOptions; +using streamr::trackerlessnetwork::contentdeliverymanagerevents::NewMessage; +using streamr::trackerlessnetwork::controllayer:: + createStreamPartDiscoveryLayerNode; +using streamr::trackerlessnetwork::controllayer::DhtNodeControlLayer; +using streamr::utils::BinaryUtils; +using streamr::utils::blockingWait; +using streamr::utils::StreamPartID; +using streamr::utils::StreamPartIDUtils; +using streamr::utils::waitForCondition; + +namespace { + +constexpr std::chrono::milliseconds neighborUpdateInterval{100}; +constexpr std::chrono::seconds untilTimeout{15}; +constexpr std::chrono::milliseconds pollInterval{100}; + +// Local copies of the TestUtils factories (see the NB above). +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(666); // NOLINT + 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; +} + +} // namespace + +class ContentDeliveryManagerTest : public ::testing::Test { +protected: + PeerDescriptor peerDescriptor1 = createMockPeerDescriptor(); + PeerDescriptor peerDescriptor2 = createMockPeerDescriptor(); + StreamPartID streamPartId = StreamPartIDUtils::parse("test#0"); + Simulator simulator{LatencyType::NONE}; + std::shared_ptr transport1; + std::shared_ptr transport2; + std::shared_ptr controlLayerNode1; + std::shared_ptr controlLayerNode2; + std::shared_ptr manager1; + std::shared_ptr manager2; + + void SetUp() override { + this->transport1 = std::make_shared( + this->peerDescriptor1, this->simulator); + this->transport1->start(); + this->transport2 = std::make_shared( + this->peerDescriptor2, this->simulator); + this->transport2->start(); + this->controlLayerNode1 = std::make_shared( + std::make_shared(DhtNodeOptions{ + .transport = this->transport1.get(), + .connectionsView = this->transport1.get(), + .peerDescriptor = this->peerDescriptor1, + .entryPoints = {this->peerDescriptor1}})); + this->controlLayerNode2 = std::make_shared( + std::make_shared(DhtNodeOptions{ + .transport = this->transport2.get(), + .connectionsView = this->transport2.get(), + .peerDescriptor = this->peerDescriptor2, + .entryPoints = {this->peerDescriptor1}})); + blockingWait( + folly::coro::collectAll( + this->controlLayerNode1->start(), + this->controlLayerNode2->start())); + blockingWait( + folly::coro::collectAll( + this->controlLayerNode1->joinDht({this->peerDescriptor1}), + this->controlLayerNode2->joinDht({this->peerDescriptor1}))); + + this->manager1 = std::make_shared( + ContentDeliveryManagerOptions{ + .neighborUpdateInterval = neighborUpdateInterval, + .createDiscoveryLayerNode = + [this]( + const StreamPartID& id, + std::vector entryPoints) { + return createStreamPartDiscoveryLayerNode( + id, + std::move(entryPoints), + *this->controlLayerNode1); + }}); + this->manager2 = std::make_shared( + ContentDeliveryManagerOptions{ + .neighborUpdateInterval = neighborUpdateInterval, + .createDiscoveryLayerNode = + [this]( + const StreamPartID& id, + std::vector entryPoints) { + return createStreamPartDiscoveryLayerNode( + id, + std::move(entryPoints), + *this->controlLayerNode2); + }}); + blockingWait(this->manager1->start( + *this->controlLayerNode1, *this->transport1, *this->transport1)); + this->manager1->setStreamPartEntryPoints( + this->streamPartId, {this->peerDescriptor1}); + blockingWait(this->manager2->start( + *this->controlLayerNode2, *this->transport2, *this->transport2)); + this->manager2->setStreamPartEntryPoints( + this->streamPartId, {this->peerDescriptor1}); + } + + void TearDown() override { + blockingWait(this->manager1->destroy()); + blockingWait(this->manager2->destroy()); + blockingWait(this->controlLayerNode1->stop()); + blockingWait(this->controlLayerNode2->stop()); + this->transport1->stop(); + this->transport2->stop(); + this->simulator.stop(); + } + + void joinBothAndAwaitNeighbors() { + this->manager1->joinStreamPart(this->streamPartId); + this->manager2->joinStreamPart(this->streamPartId); + blockingWait(waitForCondition( + [this]() { + return this->manager1->getNeighbors(this->streamPartId) + .size() == 1 && + this->manager2->getNeighbors(this->streamPartId).size() == + 1; + }, + untilTimeout, + pollInterval)); + } +}; + +TEST_F(ContentDeliveryManagerTest, Starts) { + EXPECT_EQ( + Identifiers::getNodeIdFromPeerDescriptor( + this->manager1->getPeerDescriptor()), + Identifiers::getNodeIdFromPeerDescriptor(this->peerDescriptor1)); + EXPECT_EQ( + Identifiers::getNodeIdFromPeerDescriptor( + this->manager2->getPeerDescriptor()), + Identifiers::getNodeIdFromPeerDescriptor(this->peerDescriptor2)); +} + +TEST_F(ContentDeliveryManagerTest, JoiningStream) { + this->joinBothAndAwaitNeighbors(); + EXPECT_EQ(this->manager1->getNeighbors(this->streamPartId).size(), 1); + EXPECT_EQ(this->manager2->getNeighbors(this->streamPartId).size(), 1); +} + +TEST_F( + ContentDeliveryManagerTest, PublishingAfterJoiningAndWaitingForNeighbors) { + this->joinBothAndAwaitNeighbors(); + std::atomic received1 = 0; + this->manager1->on( + [&received1](const StreamMessage& /*msg*/) { received1++; }); + this->manager2->broadcast( + createLocalStreamMessage(R"({"hello":"WORLD"})", this->streamPartId)); + blockingWait(waitForCondition( + [&received1]() { return received1 >= 1; }, untilTimeout, pollInterval)); +} + +TEST_F(ContentDeliveryManagerTest, MultiStreamPubSub) { + const auto streamPartId2 = StreamPartIDUtils::parse("test2#0"); + this->manager1->setStreamPartEntryPoints( + streamPartId2, {this->peerDescriptor1}); + this->manager2->setStreamPartEntryPoints( + streamPartId2, {this->peerDescriptor1}); + this->manager1->joinStreamPart(this->streamPartId); + this->manager1->joinStreamPart(streamPartId2); + this->manager2->joinStreamPart(this->streamPartId); + this->manager2->joinStreamPart(streamPartId2); + blockingWait(waitForCondition( + [this, &streamPartId2]() { + return this->manager1->getNeighbors(this->streamPartId).size() == + 1 && + this->manager2->getNeighbors(this->streamPartId).size() == 1 && + this->manager1->getNeighbors(streamPartId2).size() == 1 && + this->manager2->getNeighbors(streamPartId2).size() == 1; + }, + untilTimeout, + pollInterval)); + std::atomic received1 = 0; + std::atomic received2 = 0; + this->manager1->on( + [&received1](const StreamMessage& /*msg*/) { received1++; }); + this->manager2->on( + [&received2](const StreamMessage& /*msg*/) { received2++; }); + this->manager1->broadcast( + createLocalStreamMessage(R"({"hello":"WORLD"})", streamPartId2)); + this->manager2->broadcast( + createLocalStreamMessage(R"({"hello":"WORLD"})", this->streamPartId)); + blockingWait(waitForCondition( + [&received1, &received2]() { return received1 >= 1 && received2 >= 1; }, + untilTimeout, + pollInterval)); +} + +TEST_F(ContentDeliveryManagerTest, LeavingStreamParts) { + this->joinBothAndAwaitNeighbors(); + blockingWait(this->manager2->leaveStreamPart(this->streamPartId)); + blockingWait(waitForCondition( + [this]() { + return this->manager1->getNeighbors(this->streamPartId).empty(); + }, + untilTimeout, + pollInterval)); +} + +TEST_F(ContentDeliveryManagerTest, RttsAreUpdatedForNodeInfo) { + this->joinBothAndAwaitNeighbors(); + // Wait for the neighbor-update round to record RTTs. + blockingWait(waitForCondition( + [this]() { + const auto info1 = this->manager1->getNodeInfo(); + const auto info2 = this->manager2->getNodeInfo(); + return !info1.empty() && + !info1[0].contentDeliveryLayerNeighbors.empty() && + info1[0].contentDeliveryLayerNeighbors[0].rtt.has_value() && + !info2.empty() && + !info2[0].contentDeliveryLayerNeighbors.empty() && + info2[0].contentDeliveryLayerNeighbors[0].rtt.has_value(); + }, + untilTimeout, + pollInterval)); + const auto nodeInfo1 = this->manager1->getNodeInfo(); + EXPECT_GE(nodeInfo1[0].contentDeliveryLayerNeighbors[0].rtt.value(), 0); +}