diff --git a/packages/streamr-dht/modules/dht/DhtNode.cppm b/packages/streamr-dht/modules/dht/DhtNode.cppm index a1326fea..6d78b4cb 100644 --- a/packages/streamr-dht/modules/dht/DhtNode.cppm +++ b/packages/streamr-dht/modules/dht/DhtNode.cppm @@ -150,6 +150,10 @@ struct DhtNodeOptions { std::optional peerDescriptor; std::vector entryPoints; + // Whether remote peers may mark their connection to us as private + // (proxy clients do); passed through to the owned ConnectionManager. + // TS NetworkStack sets this from acceptProxyConnections. + bool allowIncomingPrivateConnections = false; // Connectivity options for the owned-ConnectionManager path (TS // DhtNodeOptions). websocketServerEnableTls/TLS certificates and the // GeoIP region fallback stay deferred (milestones D/E); region defaults @@ -551,7 +555,9 @@ public: -> std::shared_ptr { return std::make_shared( facadeOptions); - }}); + }, + .allowIncomingPrivateConnections = + this->options.allowIncomingPrivateConnections}); this->ownedConnectionManager->start(); this->transportPtr = this->ownedConnectionManager.get(); this->connectionsView = this->ownedConnectionManager.get(); @@ -928,6 +934,14 @@ public: return this->peerDiscovery->isJoinCalled(); } + // The transport the node runs on (the owned ConnectionManager on + // the no-transport-given path); TS DhtNode.getTransport(). + [[nodiscard]] Transport* getTransport() { return this->transportPtr; } + + [[nodiscard]] ConnectionLocker* getConnectionLocker() { + return this->connectionLocker; + } + [[nodiscard]] ConnectionsView* getConnectionsView() { return this->connectionsView; } diff --git a/packages/streamr-dht/modules/dht/recursive-operation/RecursiveOperationManager.cppm b/packages/streamr-dht/modules/dht/recursive-operation/RecursiveOperationManager.cppm index 4fefb423..279a78c6 100644 --- a/packages/streamr-dht/modules/dht/recursive-operation/RecursiveOperationManager.cppm +++ b/packages/streamr-dht/modules/dht/recursive-operation/RecursiveOperationManager.cppm @@ -17,6 +17,7 @@ module; #include #include +#include #include #include #include @@ -116,6 +117,16 @@ private: std::recursive_mutex mutex; std::map> ongoingSessions; + // Completed sessions are RETIRED here instead of dying at the end of + // execute(): a routed message for the session may still be mid-flight + // through the session communicator's transport handler on another + // thread (or the execute() continuation may run inline inside that + // very handler), and destroying the session there is a + // use-after-free. The bounded ring defers destruction by + // retiredSessionCapacity completed sessions — long past any handler + // invocation. Guarded by this->mutex. + static constexpr size_t retiredSessionCapacity = 16; + std::deque> retiredSessions; bool stopped = false; std::unique_ptr rpcLocal; // sendResponse()'s per-session communicators whose internal scopes @@ -459,6 +470,19 @@ public: {this->options.localPeerDescriptor}, dataEntries, true); + // TS returns without stopping the session; its GC keeps the + // still-subscribed transport listener alive. Here the session + // dies with this frame, so the listeners MUST be detached + // first — skipping this leaves a dangling Message handler on + // the transport (use-after-free on the next routed message). + session->stop(); + { + std::scoped_lock lock(this->mutex); + this->retiredSessions.push_back(session); + while (this->retiredSessions.size() > retiredSessionCapacity) { + this->retiredSessions.pop_front(); + } + } co_return session->getResults(); } { @@ -504,6 +528,13 @@ public: this->ongoingSessions.erase(session->getId()); } session->stop(); + { + std::scoped_lock lock(this->mutex); + this->retiredSessions.push_back(session); + while (this->retiredSessions.size() > retiredSessionCapacity) { + this->retiredSessions.pop_front(); + } + } co_return session->getResults(); } @@ -515,6 +546,7 @@ public: session->stop(); } this->ongoingSessions.clear(); + this->retiredSessions.clear(); } // Drain the detached response sends BEFORE destroying the retired // communicators: a draining task may still retire its communicator diff --git a/packages/streamr-dht/modules/dht/recursive-operation/RecursiveOperationSession.cppm b/packages/streamr-dht/modules/dht/recursive-operation/RecursiveOperationSession.cppm index 8ca53d08..ced56ab7 100644 --- a/packages/streamr-dht/modules/dht/recursive-operation/RecursiveOperationSession.cppm +++ b/packages/streamr-dht/modules/dht/recursive-operation/RecursiveOperationSession.cppm @@ -339,12 +339,20 @@ public: } void stop() { - std::scoped_lock lock(this->mutex); - this->abortController.abort(); + { + std::scoped_lock lock(this->mutex); + this->abortController.abort(); + } + // NOT under this->mutex: destroy() waits out an in-flight + // transport Message handler, and that handler (onResponseReceived) + // takes this->mutex — holding it here is an ABBA deadlock. this->rpcCommunicator.destroy(); - if (!this->completionEventEmitted) { - this->completionEventEmitted = true; - this->emit(); + { + std::scoped_lock lock(this->mutex); + if (!this->completionEventEmitted) { + this->completionEventEmitted = true; + this->emit(); + } } } }; diff --git a/packages/streamr-dht/modules/transport/ListeningRpcCommunicator.cppm b/packages/streamr-dht/modules/transport/ListeningRpcCommunicator.cppm index d73acc8e..2f97da7a 100644 --- a/packages/streamr-dht/modules/transport/ListeningRpcCommunicator.cppm +++ b/packages/streamr-dht/modules/transport/ListeningRpcCommunicator.cppm @@ -3,6 +3,7 @@ // streamr-dht/transport/ListeningRpcCommunicator.hpp (MODERNIZATION.md // Phase 2.6): this file is now the source of truth. module; +#include #include #include @@ -17,6 +18,7 @@ import streamr.protorpc.Errors; import streamr.protorpc.RpcCommunicator; import streamr.dht.RoutingRpcCommunicator; import streamr.dht.Transport; +import streamr.logger.SLogger; // Hoisted from the former header (file scope, NOT exported); // fully qualified: relative namespace names resolve differently @@ -27,6 +29,8 @@ using streamr::protorpc::RpcClientError; // neighboring header before consolidation). using streamr::protorpc::RpcCommunicatorOptions; +using streamr::logger::SLogger; + export namespace streamr::dht::transport { using ::dht::PeerDescriptor; @@ -40,6 +44,8 @@ private: HandlerToken onDisconnectedHandlerToken; public: + std::string instrServiceId; // INSTR + ListeningRpcCommunicator( ServiceID&& serviceId, Transport& transport, @@ -54,6 +60,11 @@ public: this->handleMessageFromPeer(message); }), transport(transport) { + this->instrServiceId = this->getOwnServiceId(); + SLogger::info( + "INSTR ListeningRpcCommunicator ctor", + {{"serviceId", this->instrServiceId}, + {"ptr", reinterpret_cast(this)}}); this->onMessageHandlerToken = transport.on( [this](const Message& message) { this->listener(message); }); this->onDisconnectedHandlerToken = @@ -76,7 +87,18 @@ public: }); } + ~ListeningRpcCommunicator() { + SLogger::info( + "INSTR ~ListeningRpcCommunicator", + {{"serviceId", this->instrServiceId}, + {"ptr", reinterpret_cast(this)}}); + } + void destroy() { + SLogger::info( + "INSTR ListeningRpcCommunicator::destroy", + {{"serviceId", this->instrServiceId}, + {"ptr", reinterpret_cast(this)}}); transport.off(this->onMessageHandlerToken); // Also detach the Disconnected listener: leaving it registered // dangles `this` on the transport after destruction, and a live diff --git a/packages/streamr-dht/modules/transport/RoutingRpcCommunicator.cppm b/packages/streamr-dht/modules/transport/RoutingRpcCommunicator.cppm index 56a2754a..a259789d 100644 --- a/packages/streamr-dht/modules/transport/RoutingRpcCommunicator.cppm +++ b/packages/streamr-dht/modules/transport/RoutingRpcCommunicator.cppm @@ -122,6 +122,10 @@ public: // straggler request drained that late runs against destroyed members // and a possibly-destroyed transport (observed as the Layer0 // end-to-end teardown SIGSEGV, macOS crash reports 2026-07-11). + [[nodiscard]] const ServiceID& getOwnServiceId() const { + return this->ownServiceId; + } + ~RoutingRpcCommunicator() { this->drainAsyncTasks(); } RoutingRpcCommunicator(const RoutingRpcCommunicator&) = delete; diff --git a/packages/streamr-trackerless-network/CMakeLists.txt b/packages/streamr-trackerless-network/CMakeLists.txt index 902149ae..ded21467 100644 --- a/packages/streamr-trackerless-network/CMakeLists.txt +++ b/packages/streamr-trackerless-network/CMakeLists.txt @@ -186,6 +186,14 @@ if(NOT IOS AND STREAMR_MODULES_SUPPORTED) test/unit/ContentDeliveryManagerTest.cpp test/unit/ContentDeliveryLayerNodeLayer1Test.cpp test/unit/PropagationScaleTest.cpp + test/unit/NetworkNodeTest.cpp + test/unit/NetworkRpcTest.cpp + test/unit/NetworkNodeIntegrationTest.cpp + test/unit/NetworkStackTest.cpp + test/unit/NodeInfoRpcTest.cpp + test/unit/ExternalNetworkRpcTest.cpp + test/unit/ProxyConnectionsTest.cpp + test/unit/ProxyAndFullNodeTest.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 eea747bc..ce31fb29 100755 --- a/packages/streamr-trackerless-network/lint.sh +++ b/packages/streamr-trackerless-network/lint.sh @@ -42,8 +42,13 @@ fi # unification false positive inside the generated protobuf setter # (set_content -> ArenaStringPtr::SetBytes); the compiler builds and # runs it. +# NetworkNodeTest.cpp, ProxyAndFullNodeTest.cpp and +# ProxyConnectionsTest.cpp (phase C6) trip the same std-type +# unification false positives (generated protobuf setter and +# std::string member calls on own locals); the compiler builds and +# runs all three. 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' | grep -v 'test/unit/ContentDeliveryManagerTest.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' | grep -v 'test/unit/NetworkNodeTest.cpp' | grep -v 'test/unit/ProxyAndFullNodeTest.cpp' | grep -v 'test/unit/ProxyConnectionsTest.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 index 6e4c0e30..83d8f9a9 100644 --- a/packages/streamr-trackerless-network/modules/ContentDeliveryManager.cppm +++ b/packages/streamr-trackerless-network/modules/ContentDeliveryManager.cppm @@ -124,9 +124,11 @@ struct StreamPartitionInfo { class ContentDeliveryManager : public EventEmitter { -private: +public: // The TS StreamPartDelivery union: proxied=false carries the layer - // objects, proxied=true carries the client. + // objects, proxied=true carries the client. Public: the proxy + // end-to-end tests (and diagnostics) inspect it via + // getStreamPartDelivery(). struct StreamPartDelivery { bool proxied = false; std::function broadcast; @@ -141,6 +143,7 @@ private: std::shared_ptr client; }; +private: ContentDeliveryManagerOptions options; ControlLayerNode* controlLayerNode = nullptr; Transport* transport = nullptr; @@ -187,6 +190,22 @@ public: SLogger::trace("Destroying ContentDeliveryManager"); this->destroyed = true; } + // Signal every part's split-avoidance and reconnect loops to + // abort BEFORE joining the scheduled tasks: close() joins the + // avoidNetworkSplit/reconnect coroutines, and without the abort + // they run their full exponential backoff (~a minute per part) + // before finishing. + { + std::scoped_lock lock(this->mutex); + for (const auto& [id, part] : this->streamParts) { + if (part->networkSplitAvoidance) { + part->networkSplitAvoidance->destroy(); + } + if (part->streamPartReconnect) { + part->streamPartReconnect->destroy(); + } + } + } this->joinScope.close(); std::vector> parts; { @@ -359,7 +378,7 @@ public: folly::coro::Task setProxies( StreamPartID streamPartId, std::vector nodes, - ProxyDirection direction, + std::optional direction, EthereumAddress userId, std::optional connectionCount = std::nullopt) { // TS TODO preserved: explicit default value for @@ -465,6 +484,13 @@ public: it->second->client->getDirection() == direction.value()); } + [[nodiscard]] std::shared_ptr getStreamPartDelivery( + const StreamPartID& streamPartId) const { + std::scoped_lock lock(this->mutex); + const auto it = this->streamParts.find(streamPartId); + return it != this->streamParts.end() ? it->second : nullptr; + } + [[nodiscard]] bool hasStreamPart(const StreamPartID& streamPartId) const { std::scoped_lock lock(this->mutex); return this->streamParts.contains(streamPartId); diff --git a/packages/streamr-trackerless-network/modules/NetworkNode.cppm b/packages/streamr-trackerless-network/modules/NetworkNode.cppm new file mode 100644 index 00000000..03594612 --- /dev/null +++ b/packages/streamr-trackerless-network/modules/NetworkNode.cppm @@ -0,0 +1,193 @@ +// Module streamr.trackerlessnetwork.NetworkNode +// Ported from packages/trackerless-network/src/NetworkNode.ts +// (v103.8.0-rc.3): the public client-facing API over a NetworkStack. +// +// Adaptations: addMessageListener returns a HandlerToken and +// removeMessageListener takes it (the C++ EventEmitter unsubscribes by +// token, not listener identity). getMetricsContext/getDiagnosticInfo +// are not ported (consistent with earlier phases). The TS generic +// registerExternalNetworkRpcMethod/createExternalRpcClient class-object +// parameters are template parameters. +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 + +export module streamr.trackerlessnetwork.NetworkNode; + +import streamr.utils.CoroutineHelper; +import streamr.utils.EthereumAddress; +import streamr.utils.StreamPartID; +import streamr.trackerlessnetwork.protos; +import streamr.trackerlessnetwork.ContentDeliveryManager; +import streamr.trackerlessnetwork.ExternalNetworkRpc; +import streamr.trackerlessnetwork.NetworkStack; +import streamr.dht.Identifiers; +import streamr.dht.protos; +import streamr.eventemitter.EventEmitter; + +using streamr::dht::DhtAddress; +using streamr::eventemitter::HandlerToken; +using streamr::utils::EthereumAddress; +using streamr::utils::StreamPartID; + +export namespace streamr::trackerlessnetwork { + +using ::dht::PeerDescriptor; +using streamr::trackerlessnetwork::controllayer::ExternalNetworkRpc; + +/** + * Convenience wrapper for building client-facing functionality (TS + * comment: used by client). + */ +class NetworkNode { +private: + std::shared_ptr stack; + std::unique_ptr externalNetworkRpc; + bool stopped = false; + +public: + explicit NetworkNode(std::shared_ptr stack) + : stack(std::move(stack)) {} + + folly::coro::Task start(bool doJoin = true) { + co_await this->stack->start(doJoin); + this->externalNetworkRpc = + std::make_unique(*this->stack->getTransport()); + } + + folly::coro::Task inspect( + PeerDescriptor node, StreamPartID streamPartId) { + co_return co_await this->stack->getContentDeliveryManager().inspect( + std::move(node), std::move(streamPartId)); + } + + folly::coro::Task broadcast(const StreamMessage& msg) { + co_await this->stack->broadcast(msg); + } + + folly::coro::Task join( + StreamPartID streamPartId, + std::optional neighborRequirement = std::nullopt) { + co_await this->stack->joinStreamPart( + std::move(streamPartId), neighborRequirement); + } + + folly::coro::Task setProxies( + StreamPartID streamPartId, + std::vector nodes, + std::optional direction, + const EthereumAddress& userId, + std::optional connectionCount = std::nullopt) { + co_await this->stack->getContentDeliveryManager().setProxies( + std::move(streamPartId), + std::move(nodes), + direction, + userId, + connectionCount); + } + + [[nodiscard]] bool isProxiedStreamPart( + const StreamPartID& streamPartId) const { + return this->stack->getContentDeliveryManager().isProxiedStreamPart( + streamPartId); + } + + // The C++ EventEmitter unsubscribes by token: keep the returned + // token and pass it to removeMessageListener (TS removes by + // listener identity). + HandlerToken addMessageListener( + const std::function& listener) { + return this->stack->getContentDeliveryManager() + .on(listener); + } + + void removeMessageListener(HandlerToken token) { + this->stack->getContentDeliveryManager() + .off(token); + } + + void setStreamPartEntryPoints( + const StreamPartID& streamPartId, + std::vector contactPeerDescriptors) { + this->stack->getContentDeliveryManager().setStreamPartEntryPoints( + streamPartId, std::move(contactPeerDescriptors)); + } + + folly::coro::Task leave(StreamPartID streamPartId) { + if (this->stopped) { + co_return; + } + co_await this->stack->getContentDeliveryManager().leaveStreamPart( + std::move(streamPartId)); + } + + [[nodiscard]] std::vector getNeighbors( + const StreamPartID& streamPartId) const { + return this->stack->getContentDeliveryManager().getNeighbors( + streamPartId); + } + + [[nodiscard]] bool hasStreamPart(const StreamPartID& streamPartId) const { + return this->stack->getContentDeliveryManager().hasStreamPart( + streamPartId); + } + + folly::coro::Task stop() { + this->stopped = true; + if (this->externalNetworkRpc) { + this->externalNetworkRpc->destroy(); + } + co_await this->stack->stop(); + } + + [[nodiscard]] PeerDescriptor getPeerDescriptor() { + return this->stack->getControlLayerNode().getLocalPeerDescriptor(); + } + + [[nodiscard]] DhtAddress getNodeId() const { + return this->stack->getContentDeliveryManager().getNodeId(); + } + + [[nodiscard]] const NetworkOptions& getOptions() const { + return this->stack->getOptions(); + } + + [[nodiscard]] std::vector getStreamParts() const { + return this->stack->getContentDeliveryManager().getStreamParts(); + } + + folly::coro::Task fetchNodeInfo(PeerDescriptor node) { + co_return co_await this->stack->fetchNodeInfo(std::move(node)); + } + + template + void registerExternalNetworkRpcMethod(const std::string& name, F&& fn) { + this->externalNetworkRpc->registerRpcMethod( + name, std::forward(fn)); + } + + template + [[nodiscard]] ClientType createExternalRpcClient() { + return this->externalNetworkRpc->createRpcClient(); + } + + // TS exposes the stack as a readonly field; the proxy end-to-end + // tests reach through it. + [[nodiscard]] NetworkStack& getStack() { return *this->stack; } +}; + +inline std::shared_ptr createNetworkNode(NetworkOptions options) { + return std::make_shared( + std::make_shared(std::move(options))); +} + +} // namespace streamr::trackerlessnetwork diff --git a/packages/streamr-trackerless-network/modules/NetworkStack.cppm b/packages/streamr-trackerless-network/modules/NetworkStack.cppm new file mode 100644 index 00000000..fe998cab --- /dev/null +++ b/packages/streamr-trackerless-network/modules/NetworkStack.cppm @@ -0,0 +1,341 @@ +// Module streamr.trackerlessnetwork.NetworkStack +// Ported from packages/trackerless-network/src/NetworkStack.ts +// (v103.8.0-rc.3): the layer-0 DhtNode + ContentDeliveryManager +// composition with lifecycle and the node-info RPC service. +// +// Adaptations: the TS module-level instances list and process/window +// exit handlers are a Node.js/browser runtime concern and are not +// ported — C++ callers own the stack's lifetime (the shared library has +// its own teardown). MetricsContext is not ported (consistent with +// earlier phases). The TS setImmediate() fire-and-forget joinDht runs +// as a bounded GuardedAsyncScope task. The manager's discovery-layer +// factory (injected in C5 because the DhtNode module graph cannot be +// composed inside ContentDeliveryManager.cppm) is supplied here with +// createStreamPartDiscoveryLayerNode, matching the TS inline +// construction. +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 + +export module streamr.trackerlessnetwork.NetworkStack; + +import streamr.utils.CoroutineHelper; +import streamr.utils.GuardedAsyncScope; +import streamr.utils.SharedExecutors; +import streamr.utils.StreamID; +import streamr.utils.StreamPartID; +import streamr.utils.waitForCondition; +import streamr.trackerlessnetwork.protos; +import streamr.trackerlessnetwork.ContentDeliveryManager; +import streamr.trackerlessnetwork.ControlLayerNode; +import streamr.trackerlessnetwork.DhtNodeControlLayer; +import streamr.trackerlessnetwork.createStreamPartDiscoveryLayerNode; +import streamr.trackerlessnetwork.NodeInfoClient; +import streamr.trackerlessnetwork.NodeInfoRpcLocal; +import streamr.dht.ConnectionLocker; +import streamr.dht.DhtNode; +import streamr.dht.Identifiers; +import streamr.dht.ListeningRpcCommunicator; +import streamr.dht.Transport; +import streamr.dht.Version; +import streamr.dht.protos; +import streamr.eventemitter.EventEmitter; +import streamr.logger.SLogger; + +// Hoisted (file scope, NOT exported); fully qualified because relative +// namespace names resolve differently at file scope than inside the +// package namespace. +using streamr::dht::DhtNode; +using streamr::dht::Identifiers; +using streamr::dht::ServiceID; +using streamr::dht::connection::ConnectionLocker; +using streamr::dht::helpers::Version; +using streamr::dht::transport::ListeningRpcCommunicator; +using streamr::dht::transport::Transport; +using streamr::logger::SLogger; +using streamr::utils::GuardedAsyncScope; +using streamr::utils::StreamID; +using streamr::utils::StreamPartID; +using streamr::utils::waitForCondition; + +export namespace streamr::trackerlessnetwork { + +using ::dht::PeerDescriptor; +using streamr::trackerlessnetwork::controllayer::ControlLayerNode; +using streamr::trackerlessnetwork::controllayer:: + createStreamPartDiscoveryLayerNode; +using streamr::trackerlessnetwork::controllayer::DhtNodeControlLayer; + +// TS NetworkOptions, minus metricsContext (not ported). +struct NetworkOptions { + streamr::dht::DhtNodeOptions layer0; + ContentDeliveryManagerOptions networkNode; +}; + +// TS joinStreamPart neighborRequirement parameter. +struct NeighborRequirement { + size_t minCount; + std::chrono::milliseconds timeout; +}; + +class NetworkStack { +private: + NetworkOptions options; + std::shared_ptr dhtNode; + std::shared_ptr controlLayerNode; + std::shared_ptr contentDeliveryManager; + std::unique_ptr infoRpcCommunicator; + std::unique_ptr nodeInfoRpcLocal; + std::unique_ptr nodeInfoClient; + bool stopped = false; + // The TS setImmediate() joinDht; bounded by the dht join timeout. + streamr::utils::SharedSerialExecutor joinExecutor{ + streamr::utils::SharedExecutors::worker()}; + GuardedAsyncScope joinScope; + + folly::coro::Task ensureConnectedToControlLayer() { + // TS TODO preserved: could wrap joinDht with pOnce and call it + // here. + if (!this->controlLayerNode->hasJoined()) { + if (!this->options.layer0.entryPoints.empty()) { + // TS TODO preserved: should catch possible rejection? + this->joinScope.add( + streamr::utils::co_withExecutor( + &this->joinExecutor, + folly::coro::co_invoke( + [this]() -> folly::coro::Task { + co_await this->controlLayerNode->joinDht( + this->options.layer0.entryPoints); + }))); + } + co_await this->controlLayerNode->waitForNetworkConnectivity(); + } + } + +public: + explicit NetworkStack(NetworkOptions options) + : options(std::move(options)) { + auto layer0 = this->options.layer0; + // TS: allowIncomingPrivateConnections = + // options.networkNode?.acceptProxyConnections. + layer0.allowIncomingPrivateConnections = + this->options.networkNode.acceptProxyConnections.value_or(false); + this->dhtNode = std::make_shared(std::move(layer0)); + this->controlLayerNode = + std::make_shared(this->dhtNode); + auto managerOptions = this->options.networkNode; + if (!managerOptions.createDiscoveryLayerNode) { + // The TS inline layer-1 DhtNode construction; see the module + // comment for why the manager takes this as a factory. + managerOptions.createDiscoveryLayerNode = + [this]( + const StreamPartID& streamPartId, + std::vector entryPoints) { + return createStreamPartDiscoveryLayerNode( + streamPartId, + std::move(entryPoints), + *this->controlLayerNode); + }; + } + this->contentDeliveryManager = + std::make_shared(std::move(managerOptions)); + } + + NetworkStack(const NetworkStack&) = delete; + NetworkStack& operator=(const NetworkStack&) = delete; + NetworkStack(NetworkStack&&) = delete; + NetworkStack& operator=(NetworkStack&&) = delete; + + virtual ~NetworkStack() { streamr::utils::blockingWait(this->stop()); } + + folly::coro::Task start(bool doJoin = true) { + SLogger::info("Starting a Streamr Network Node"); + co_await this->controlLayerNode->start(); + SLogger::info( + "Node id is " + + Identifiers::getNodeIdFromPeerDescriptor( + this->controlLayerNode->getLocalPeerDescriptor())); + const auto localPeerDescriptor = + this->controlLayerNode->getLocalPeerDescriptor(); + bool localIsEntryPoint = false; + for (const auto& entryPoint : this->options.layer0.entryPoints) { + if (Identifiers::areEqualPeerDescriptors( + entryPoint, localPeerDescriptor)) { + localIsEntryPoint = true; + break; + } + } + if (localIsEntryPoint) { + co_await this->controlLayerNode->joinDht( + this->options.layer0.entryPoints); + } else if (doJoin) { + // In practice there aren't existing connections and + // therefore this always connects (TS comment). + co_await this->ensureConnectedToControlLayer(); + } + // TS casts the transport to ConnectionManager and passes it as + // both the transport and the locker; here the locker is either + // the one the node resolved (owned ConnectionManager or the + // explicitly given one) or the transport itself (the simulator + // transport implements ConnectionLocker). + auto* transport = this->dhtNode->getTransport(); + auto* connectionLocker = this->dhtNode->getConnectionLocker(); + if (connectionLocker == nullptr) { + connectionLocker = dynamic_cast(transport); + } + if (connectionLocker == nullptr) { + throw std::runtime_error( + "NetworkStack: the layer-0 transport does not provide a ConnectionLocker"); + } + co_await this->contentDeliveryManager->start( + *this->controlLayerNode, *transport, *connectionLocker); + this->infoRpcCommunicator = std::make_unique( + ServiceID{nodeInfoRpcServiceId}, *this->dhtNode->getTransport()); + this->nodeInfoRpcLocal = std::make_unique( + [this]() { return this->createNodeInfo(); }, + *this->infoRpcCommunicator); + this->nodeInfoClient = std::make_unique( + this->controlLayerNode->getLocalPeerDescriptor(), + *this->infoRpcCommunicator); + } + + // virtual so the NetworkNode unit test can stub the stack (the TS + // test passes a partial object). + virtual folly::coro::Task joinStreamPart( + StreamPartID streamPartId, + std::optional neighborRequirement = std::nullopt) { + if (this->getContentDeliveryManager().isProxiedStreamPart( + streamPartId)) { + throw std::runtime_error( + "Cannot join to " + streamPartId + + " as proxy connections have been set"); + } + co_await this->ensureConnectedToControlLayer(); + this->getContentDeliveryManager().joinStreamPart(streamPartId); + if (neighborRequirement.has_value()) { + co_await waitForCondition( + [this, streamPartId, &neighborRequirement]() { + return this->getContentDeliveryManager() + .getNeighbors(streamPartId) + .size() >= neighborRequirement->minCount; + }, + neighborRequirement->timeout); + } + } + + folly::coro::Task broadcast(const StreamMessage& msg) { + const auto streamPartId = streamr::utils::toStreamPartID( + StreamID{msg.messageid().streamid()}, + static_cast(msg.messageid().streampartition())); + if (this->getContentDeliveryManager().isProxiedStreamPart( + streamPartId, ProxyDirection::SUBSCRIBE) && + msg.has_contentmessage()) { + throw std::runtime_error( + "Cannot broadcast to " + streamPartId + + " as proxy subscribe connections have been set"); + } + // TS TODO preserved: could combine these two calls to + // isProxiedStreamPart? + if (!this->contentDeliveryManager->isProxiedStreamPart(streamPartId)) { + co_await this->ensureConnectedToControlLayer(); + } + this->getContentDeliveryManager().broadcast(msg); + } + + virtual ContentDeliveryManager& getContentDeliveryManager() { + return *this->contentDeliveryManager; + } + + [[nodiscard]] ControlLayerNode& getControlLayerNode() { + return *this->controlLayerNode; + } + + // The transport the layer-0 node runs on (TS + // controlLayerNode.getTransport(), a ConnectionManager on the + // own-connection-manager path). + [[nodiscard]] Transport* getTransport() { + return this->dhtNode->getTransport(); + } + + folly::coro::Task fetchNodeInfo(PeerDescriptor node) { + if (!Identifiers::areEqualPeerDescriptors( + node, this->getControlLayerNode().getLocalPeerDescriptor())) { + co_return co_await this->nodeInfoClient->getInfo(std::move(node)); + } + co_return this->createNodeInfo(); + } + + NodeInfoResponse createNodeInfo() { + NodeInfoResponse response; + *response.mutable_peerdescriptor() = + this->getControlLayerNode().getLocalPeerDescriptor(); + auto* controlLayer = response.mutable_controllayer(); + for (const auto& connection : this->getControlLayerNode() + .getConnectionsView() + ->getConnections()) { + *controlLayer->add_connections() = connection; + } + for (const auto& neighbor : + this->getControlLayerNode().getNeighbors()) { + *controlLayer->add_neighbors() = neighbor; + } + for (const auto& info : + this->getContentDeliveryManager().getNodeInfo()) { + auto* partition = response.add_streampartitions(); + partition->set_id(info.id); + for (const auto& neighbor : info.controlLayerNeighbors) { + *partition->add_controllayerneighbors() = neighbor; + } + for (const auto& neighbor : info.contentDeliveryLayerNeighbors) { + auto* neighborInfo = + partition->add_contentdeliverylayerneighbors(); + *neighborInfo->mutable_peerdescriptor() = + neighbor.peerDescriptor; + if (neighbor.rtt.has_value()) { + neighborInfo->set_rtt( + static_cast(neighbor.rtt.value())); + } + } + } + // TS reports the trackerless-network package version; the C++ + // port reports the SDK application version. + response.set_applicationversion(Version::localApplicationVersion); + return response; + } + + [[nodiscard]] const NetworkOptions& getOptions() const { + return this->options; + } + + folly::coro::Task stop() { + if (!this->stopped) { + this->stopped = true; + // Drain the fire-and-forget joins before the members die + // (teardown ordering; see ContentDeliveryManager::destroy). + this->joinScope.close(); + co_await this->contentDeliveryManager->destroy(); + // The info communicator listens on the transport — take it + // down while the transport is still alive. + if (this->infoRpcCommunicator) { + this->infoRpcCommunicator->destroy(); + } + this->nodeInfoRpcLocal = nullptr; + this->nodeInfoClient = nullptr; + this->infoRpcCommunicator = nullptr; + co_await this->controlLayerNode->stop(); + } + } +}; + +} // namespace streamr::trackerlessnetwork diff --git a/packages/streamr-trackerless-network/modules/NodeInfoClient.cppm b/packages/streamr-trackerless-network/modules/NodeInfoClient.cppm new file mode 100644 index 00000000..4541dee6 --- /dev/null +++ b/packages/streamr-trackerless-network/modules/NodeInfoClient.cppm @@ -0,0 +1,47 @@ +// Module streamr.trackerlessnetwork.NodeInfoClient +// Ported from packages/trackerless-network/src/NodeInfoClient.ts +// (v103.8.0-rc.3): fetches NodeInfo from a remote node over the +// node-info RPC service. +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.NodeInfoClient; + +import streamr.utils.CoroutineHelper; +import streamr.trackerlessnetwork.protos; +import streamr.trackerlessnetwork.NodeInfoRpcRemote; +import streamr.dht.DhtCallContext; +import streamr.dht.ListeningRpcCommunicator; +import streamr.dht.protos; + +using streamr::dht::rpcprotocol::DhtCallContext; +using streamr::dht::transport::ListeningRpcCommunicator; + +export namespace streamr::trackerlessnetwork { + +class NodeInfoClient { +private: + PeerDescriptor ownPeerDescriptor; + ListeningRpcCommunicator& rpcCommunicator; + +public: + NodeInfoClient( + PeerDescriptor ownPeerDescriptor, // NOLINT + ListeningRpcCommunicator& rpcCommunicator) + : ownPeerDescriptor(std::move(ownPeerDescriptor)), + rpcCommunicator(rpcCommunicator) {} + + folly::coro::Task getInfo(PeerDescriptor node) { + NodeInfoRpcClient client{this->rpcCommunicator}; + NodeInfoRpcRemote remote( + this->ownPeerDescriptor, std::move(node), std::move(client)); + co_return co_await remote.getInfo(); + } +}; + +} // namespace streamr::trackerlessnetwork diff --git a/packages/streamr-trackerless-network/modules/NodeInfoRpcLocal.cppm b/packages/streamr-trackerless-network/modules/NodeInfoRpcLocal.cppm new file mode 100644 index 00000000..ce88157e --- /dev/null +++ b/packages/streamr-trackerless-network/modules/NodeInfoRpcLocal.cppm @@ -0,0 +1,57 @@ +// Module streamr.trackerlessnetwork.NodeInfoRpcLocal +// Ported from packages/trackerless-network/src/NodeInfoRpcLocal.ts +// (v103.8.0-rc.3): serves getInfo over the node-info RPC service. +// The TS class holds the NetworkStack and calls stack.createNodeInfo(); +// here the response factory is injected as a std::function so this +// module does not import the NetworkStack module (which imports this +// one). +module; + +#include +#include +#include + +export module streamr.trackerlessnetwork.NodeInfoRpcLocal; + +import streamr.trackerlessnetwork.protos; +import streamr.trackerlessnetwork.NetworkRpcServer; +import streamr.dht.DhtCallContext; +import streamr.dht.ListeningRpcCommunicator; + +using streamr::dht::rpcprotocol::DhtCallContext; +using streamr::dht::transport::ListeningRpcCommunicator; + +export namespace streamr::trackerlessnetwork { + +// TS NODE_INFO_RPC_SERVICE_ID. +inline constexpr auto nodeInfoRpcServiceId = "system/node-info-rpc"; + +class NodeInfoRpcLocal : public streamr::protorpc::NodeInfoRpc { +private: + std::function createNodeInfo; + ListeningRpcCommunicator& rpcCommunicator; + +public: + NodeInfoRpcLocal( + std::function createNodeInfo, // NOLINT + ListeningRpcCommunicator& rpcCommunicator) + : createNodeInfo(std::move(createNodeInfo)), + rpcCommunicator(rpcCommunicator) { + this->rpcCommunicator + .registerRpcMethod( + "getInfo", + [this]( + const NodeInfoRequest& request, + const DhtCallContext& context) { + return this->getInfo(request, context); + }); + } + + NodeInfoResponse getInfo( + const NodeInfoRequest& /* request */, + const DhtCallContext& /* callContext */) override { + return this->createNodeInfo(); + } +}; + +} // namespace streamr::trackerlessnetwork diff --git a/packages/streamr-trackerless-network/modules/NodeInfoRpcRemote.cppm b/packages/streamr-trackerless-network/modules/NodeInfoRpcRemote.cppm new file mode 100644 index 00000000..41546b84 --- /dev/null +++ b/packages/streamr-trackerless-network/modules/NodeInfoRpcRemote.cppm @@ -0,0 +1,51 @@ +// Module streamr.trackerlessnetwork.NodeInfoRpcRemote +// Ported from packages/trackerless-network/src/NodeInfoRpcRemote.ts +// (v103.8.0-rc.3): the client-side wrapper for the NodeInfoRpc service. +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 + +export module streamr.trackerlessnetwork.NodeInfoRpcRemote; + +import streamr.utils.CoroutineHelper; +import streamr.trackerlessnetwork.protos; +import streamr.trackerlessnetwork.NetworkRpcClient; +import streamr.dht.DhtCallContext; +import streamr.dht.RpcRemote; +import streamr.dht.protos; + +using streamr::dht::contact::RpcRemote; +using streamr::dht::rpcprotocol::DhtCallContext; + +export namespace streamr::trackerlessnetwork { + +using ::dht::PeerDescriptor; +using NodeInfoRpcClient = streamr::protorpc::NodeInfoRpcClient; + +class NodeInfoRpcRemote : public RpcRemote { +public: + NodeInfoRpcRemote( + PeerDescriptor localPeerDescriptor, // NOLINT + PeerDescriptor remotePeerDescriptor, + NodeInfoRpcClient client, + std::optional timeout = std::nullopt) + : RpcRemote( + std::move(localPeerDescriptor), + std::move(remotePeerDescriptor), + std::move(client), + timeout) {} + + folly::coro::Task getInfo() { + auto options = this->formDhtRpcOptions({}); + co_return co_await this->getClient().getInfo( + NodeInfoRequest{}, std::move(options)); + } +}; + +} // namespace streamr::trackerlessnetwork diff --git a/packages/streamr-trackerless-network/modules/control-layer/ExternalNetworkRpc.cppm b/packages/streamr-trackerless-network/modules/control-layer/ExternalNetworkRpc.cppm new file mode 100644 index 00000000..781c63b4 --- /dev/null +++ b/packages/streamr-trackerless-network/modules/control-layer/ExternalNetworkRpc.cppm @@ -0,0 +1,62 @@ +// Module streamr.trackerlessnetwork.ExternalNetworkRpc +// Ported from packages/trackerless-network/src/control-layer/ +// ExternalNetworkRpc.ts (v103.8.0-rc.3): a ListeningRpcCommunicator on +// the dedicated external-network service that applications use to +// register their own RPC methods and create clients for them. The TS +// class-object parameters (request/response IMessageType, client +// class) are template parameters here. +module; + +#include +#include + +export module streamr.trackerlessnetwork.ExternalNetworkRpc; + +import streamr.dht.Identifiers; +import streamr.dht.ListeningRpcCommunicator; +import streamr.dht.Transport; + +using streamr::dht::ServiceID; +using streamr::dht::transport::ListeningRpcCommunicator; +using streamr::dht::transport::Transport; + +export namespace streamr::trackerlessnetwork::controllayer { + +// TS SERVICE_ID. +inline constexpr auto externalNetworkServiceId = "external-network-service"; + +class ExternalNetworkRpc { +private: + ListeningRpcCommunicator rpcCommunicator; + +public: + explicit ExternalNetworkRpc(Transport& transport) + : rpcCommunicator(ServiceID{externalNetworkServiceId}, transport) {} + + template + void registerRpcMethod(const std::string& name, F&& fn) { + this->rpcCommunicator.registerRpcMethod( + name, std::forward(fn)); + } + + template + void registerRpcNotification(const std::string& name, F&& fn) { + this->rpcCommunicator.registerRpcNotification( + name, std::forward(fn)); + } + + // TS createRpcClient(clientClass); the generated client type (e.g. + // HandshakeRpcClient) is the template argument. + template + [[nodiscard]] ClientType createRpcClient() { + return ClientType{this->rpcCommunicator}; + } + + [[nodiscard]] ListeningRpcCommunicator& getRpcCommunicator() { + return this->rpcCommunicator; + } + + void destroy() { this->rpcCommunicator.destroy(); } +}; + +} // namespace streamr::trackerlessnetwork::controllayer diff --git a/packages/streamr-trackerless-network/modules/logic/proxy/ProxyClient.cppm b/packages/streamr-trackerless-network/modules/logic/proxy/ProxyClient.cppm index 840c7b5e..59b39755 100644 --- a/packages/streamr-trackerless-network/modules/logic/proxy/ProxyClient.cppm +++ b/packages/streamr-trackerless-network/modules/logic/proxy/ProxyClient.cppm @@ -10,9 +10,11 @@ module; #include #include +#include #include #include #include +#include #include #include #include @@ -42,7 +44,6 @@ import streamr.eventemitter.EventEmitter; import streamr.logger.SLogger; import streamr.utils.AbortController; import streamr.utils.EthereumAddress; -import streamr.utils.RetryUtils; import streamr.utils.StreamPartID; import streamr.trackerlessnetwork.ContentDeliveryRpcLocal; import streamr.trackerlessnetwork.ContentDeliveryRpcRemote; @@ -67,7 +68,6 @@ using streamr::eventemitter::EventEmitter; using streamr::logger::SLogger; using streamr::utils::AbortController; using streamr::utils::EthereumAddress; -using streamr::utils::RetryUtils; using streamr::utils::StreamPartID; using streamr::dht::Identifiers; @@ -94,7 +94,8 @@ struct ProxyClientOptions { struct ProxyDefinition { std::map nodes; size_t connectionCount; - ProxyDirection direction; + // TS parity: absent = bidirectional (accepts and publishes both ways). + std::optional direction; EthereumAddress userId; }; @@ -127,7 +128,7 @@ class ProxyClient : public EventEmitter { private: struct ProxyConnection { PeerDescriptor peerDescriptor; - ProxyDirection direction; + std::optional direction; }; struct ProxyConnectionError { @@ -141,6 +142,18 @@ private: std::map duplicateDetectors; std::optional definition; std::map connections; + // TS is single-threaded; here setProxies (API thread) races + // onNodeDisconnected (transport event threads, e.g. the rtc poll + // thread) over definition/connections/duplicateDetectors — seen as + // a use-after-free reading definition->nodes keys. The mutex guards + // STATE ONLY and is NEVER held across a blocking RPC wait: a + // transport event thread parked on it may be the very thread an + // in-flight send needs (deadlock). Snapshot under the lock, act + // outside, re-check under the lock when committing. Recursive: + // unlockConnection can emit Disconnected inline on the same thread, + // re-entering onNodeDisconnected. + mutable std::recursive_mutex mutex; + bool stopped = false; // guarded by mutex Propagation propagation; NodeList neighbors; AbortController abortController; @@ -166,6 +179,7 @@ public: [this]( const MessageID& msg, const std::optional& prev) { + std::scoped_lock lock(this->mutex); return Utils::markAndCheckDuplicate( this->duplicateDetectors, msg, prev); }, @@ -233,32 +247,33 @@ public: std::vector /* successfully connected proxies */> setProxies( const std::vector& nodes, - ProxyDirection direction, + std::optional direction, const EthereumAddress& userId, std::optional connectionCount = std::nullopt) { SLogger::trace( "Setting proxies", {{"streamPartId", this->options.streamPartId}, - {"direction", direction}, {"connectionCount", connectionCount}}); if (connectionCount.has_value() && connectionCount.value() > nodes.size()) { throw std::runtime_error( "Cannot set connectionCount above the size of the configured array of nodes"); } - std::map nodesIds; - for (const auto& node : nodes) { - nodesIds[Identifiers::getNodeIdFromPeerDescriptor(node)] = node; + { + std::scoped_lock lock(this->mutex); + std::map nodesIds; + for (const auto& node : nodes) { + nodesIds[Identifiers::getNodeIdFromPeerDescriptor(node)] = node; + } + this->definition = ProxyDefinition{ + .nodes = std::move(nodesIds), + .connectionCount = connectionCount.has_value() + ? connectionCount.value() + : nodes.size(), + .direction = direction, + .userId = userId, + }; } - this->definition = ProxyDefinition{ - .nodes = nodesIds, - .connectionCount = connectionCount.has_value() - ? connectionCount.value() - : nodes.size(), - .direction = direction, - .userId = userId, - }; - return this->updateConnections(); } @@ -266,27 +281,41 @@ public: std::vector /* connection errors */, std::vector /* successfully connected proxies */> updateConnections() { - auto invalidConnections = this->getInvalidConnections(); + std::vector invalidConnections; + { + std::scoped_lock lock(this->mutex); + invalidConnections = this->getInvalidConnections(); + } for (const auto& id : invalidConnections) { this->closeConnection(id); } std::vector errors; std::vector successfullyConnected; - auto connectionCountDiff = - this->definition->connectionCount - this->connections.size(); + // Signed on purpose: connectionCount below the current + // connection count must take the shrink branch (size_t + // subtraction would wrap). + std::int64_t connectionCountDiff = 0; + { + std::scoped_lock lock(this->mutex); + connectionCountDiff = + static_cast(this->definition->connectionCount) - + static_cast(this->connections.size()); + } if (connectionCountDiff > 0) { - auto [errs, success] = - this->openRandomConnections(connectionCountDiff); + auto [errs, success] = this->openRandomConnections( + static_cast(connectionCountDiff)); errors.insert(errors.end(), errs.begin(), errs.end()); successfullyConnected.insert( successfullyConnected.end(), success.begin(), success.end()); } else if (connectionCountDiff < 0) { - this->closeRandomConnections(-connectionCountDiff); + this->closeRandomConnections( + static_cast(-connectionCountDiff)); } return {errors, successfullyConnected}; } + // Call under this->mutex. std::vector getInvalidConnections() { return this->connections | std::views::keys | std::views::filter([this](const auto& id) { @@ -301,21 +330,30 @@ public: std::vector /* connection errors */, std::vector /* successfully connected proxies */> openRandomConnections(size_t connectionCount) { - const auto proxiesToAttempt = - this->definition->nodes | std::views::keys | - std::views::filter([this](const auto& id) { - return !this->connections.contains(id); - }) | - std::views::take(connectionCount) | std::ranges::to(); + std::vector proxiesToAttempt; + std::optional direction; + EthereumAddress userId{""}; + { + std::scoped_lock lock(this->mutex); + proxiesToAttempt = this->definition->nodes | std::views::keys | + std::views::filter([this](const auto& id) { + return !this->connections.contains(id); + }) | + std::views::take(connectionCount) | + std::ranges::to(); + direction = this->definition->direction; + userId = this->definition->userId; + } std::vector errors; std::vector successfullyConnected; for (const auto& id : proxiesToAttempt) { try { - this->attemptConnection( - id, this->definition->direction, this->definition->userId); - const auto peerDescriptor = - this->connections.at(id).peerDescriptor; - successfullyConnected.push_back(peerDescriptor); + this->attemptConnection(id, direction, userId); + std::scoped_lock lock(this->mutex); + const auto it = this->connections.find(id); + if (it != this->connections.end()) { + successfullyConnected.push_back(it->second.peerDescriptor); + } } catch (const ConnectingToProxyError& e) { errors.push_back(std::move(ConnectingToProxyError(e))); } @@ -325,9 +363,19 @@ public: void attemptConnection( const DhtAddress& nodeId, - const ProxyDirection& direction, + std::optional direction, const EthereumAddress& userId) { - const auto peerDescriptor = this->definition->nodes.at(nodeId); + PeerDescriptor peerDescriptor; + { + std::scoped_lock lock(this->mutex); + const auto it = this->definition->nodes.find(nodeId); + if (it == this->definition->nodes.end()) { + // setProxies changed the definition while this attempt + // was queued. + return; + } + peerDescriptor = it->second; + } ProxyConnectionRpcClient client{this->rpcCommunicator}; ProxyConnectionRpcRemote rpcRemote( this->options.localPeerDescriptor, peerDescriptor, client); @@ -348,10 +396,19 @@ public: this->options.connectionLocker.lockConnection( peerDescriptor, LockID{SERVICE_ID}); - this->connections.emplace( - nodeId, - ProxyConnection{ - .peerDescriptor = peerDescriptor, .direction = direction}); + { + std::scoped_lock lock(this->mutex); + if (this->stopped) { + this->options.connectionLocker.unlockConnection( + peerDescriptor, LockID{SERVICE_ID}); + return; + } + this->connections.emplace( + nodeId, + ProxyConnection{ + .peerDescriptor = peerDescriptor, + .direction = direction}); + } ContentDeliveryRpcClient client{this->rpcCommunicator}; const auto remote = std::make_shared( @@ -375,39 +432,40 @@ public: void closeRandomConnections(size_t connectionCount) { std::vector> proxiesToDisconnect; - std::ranges::sample( - this->connections, - std::back_inserter(proxiesToDisconnect), - static_cast(connectionCount), - std::mt19937{std::random_device{}()}); - + { + std::scoped_lock lock(this->mutex); + std::ranges::sample( + this->connections, + std::back_inserter(proxiesToDisconnect), + static_cast(connectionCount), + std::mt19937{std::random_device{}()}); + } for (const auto& nodeId : proxiesToDisconnect) { this->closeConnection(nodeId.first); } } void closeConnection(DhtAddress nodeId) { - if (this->connections.contains(nodeId)) { - SLogger::info( - "Close proxy connection", - {{"nodeId", nodeId}, - {"streamPartId", this->options.streamPartId}}); - const auto server = this->neighbors.get(nodeId); - if (server.has_value()) { - streamr::utils::blockingWait( - server.value()->leaveStreamPartNotice( - this->options.streamPartId, false)); + std::optional peerDescriptor; + { + std::scoped_lock lock(this->mutex); + const auto it = this->connections.find(nodeId); + if (it == this->connections.end()) { + return; } - this->removeConnection(this->connections.at(nodeId).peerDescriptor); + peerDescriptor = it->second.peerDescriptor; + this->connections.erase(it); + } + SLogger::info( + "Close proxy connection", + {{"nodeId", nodeId}, {"streamPartId", this->options.streamPartId}}); + const auto server = this->neighbors.get(nodeId); + if (server.has_value()) { + streamr::utils::blockingWait(server.value()->leaveStreamPartNotice( + this->options.streamPartId, false)); } - } - - void removeConnection(const PeerDescriptor& peerDescriptor) { - const auto nodeId = - Identifiers::getNodeIdFromPeerDescriptor(peerDescriptor); this->options.connectionLocker.unlockConnection( - peerDescriptor, LockID{SERVICE_ID}); - this->connections.erase(nodeId); + peerDescriptor.value(), LockID{SERVICE_ID}); this->neighbors.remove(nodeId); } @@ -433,8 +491,11 @@ public: SLogger::debug( "ProxyClient::broadcast() calling Utils::markAndCheckDuplicate() on message: " + msg.DebugString()); - Utils::markAndCheckDuplicate( - this->duplicateDetectors, msg.messageid(), std::nullopt); + { + std::scoped_lock lock(this->mutex); + Utils::markAndCheckDuplicate( + this->duplicateDetectors, msg.messageid(), std::nullopt); + } } SLogger::debug("ProxyClient::broadcast() emitting Message event"); this->emit(msg); @@ -460,12 +521,16 @@ public: } [[nodiscard]] bool hasConnection( - const DhtAddress& nodeId, ProxyDirection direction) const { + const DhtAddress& nodeId, + std::optional direction = std::nullopt) const { + std::scoped_lock lock(this->mutex); return this->connections.contains(nodeId) && - this->connections.at(nodeId).direction == direction; + (!direction.has_value() || + this->connections.at(nodeId).direction == direction); } - [[nodiscard]] ProxyDirection getDirection() const { + [[nodiscard]] std::optional getDirection() const { + std::scoped_lock lock(this->mutex); return this->definition->direction; } @@ -481,16 +546,22 @@ public: void onNodeDisconnected(const PeerDescriptor& peerDescriptor) { const auto nodeId = Identifiers::getNodeIdFromPeerDescriptor(peerDescriptor); - if (this->connections.contains(nodeId)) { - this->options.connectionLocker.unlockConnection( - peerDescriptor, LockID{SERVICE_ID}); - this->removeConnection(peerDescriptor); - streamr::utils::blockingWait( - RetryUtils::constantRetry( - [this]() -> void { this->updateConnections(); }, - "updating proxy connections", - this->abortController)); + { + std::scoped_lock lock(this->mutex); + if (this->stopped || this->connections.erase(nodeId) == 0) { + return; + } } + this->options.connectionLocker.unlockConnection( + peerDescriptor, LockID{SERVICE_ID}); + this->neighbors.remove(nodeId); + // Deviation from TS: no automatic reconnection attempt here. The + // TS single retry always fails in practice (the proxy that + // disconnected is gone or has left the stream part), and issuing + // blocking RPCs from transport event threads (this handler runs + // on websocket/rtc threads) deadlocks against the in-flight + // sends those threads service. Callers reconnect by re-issuing + // setProxies() — see the reconnect end-to-end test. } void start() { @@ -507,6 +578,16 @@ public: } void stop() { + { + std::scoped_lock lock(this->mutex); + if (this->stopped) { + return; + } + this->stopped = true; + this->connections.clear(); + } + // The blocking leave notices run OUTSIDE the mutex: a transport + // event thread parked on it may be the thread these sends need. auto allNeighbors = this->neighbors.getAll(); for (const auto& remote : allNeighbors) { this->options.connectionLocker.unlockConnection( @@ -517,7 +598,6 @@ public: this->neighbors.stop(); this->rpcCommunicator.destroy(); - this->connections.clear(); this->abortController.abort(); if (this->transportDisconnectedHandlerToken.has_value()) { this->options.transport diff --git a/packages/streamr-trackerless-network/modules/logic/proxy/ProxyConnectionRpcLocal.cppm b/packages/streamr-trackerless-network/modules/logic/proxy/ProxyConnectionRpcLocal.cppm index 78ce2a9b..3ccfe1a0 100644 --- a/packages/streamr-trackerless-network/modules/logic/proxy/ProxyConnectionRpcLocal.cppm +++ b/packages/streamr-trackerless-network/modules/logic/proxy/ProxyConnectionRpcLocal.cppm @@ -70,8 +70,9 @@ class ProxyConnectionRpcLocal public ProxyConnectionRpc { private: struct ProxyConnection { - ProxyDirection - direction; // Direction is from the client's point of view + // Direction is from the client's point of view; absent = + // bidirectional (TS parity). + std::optional direction; EthereumAddress userId; ContentDeliveryRpcRemote remote; }; @@ -91,7 +92,8 @@ private: std::vector getSubscribers() const { return connections | std::views::keys | std::views::filter([&](const auto& nodeId) { - return connections.at(nodeId)->direction == + return !connections.at(nodeId)->direction.has_value() || + connections.at(nodeId)->direction == ProxyDirection::SUBSCRIBE; }) | std::ranges::to(); @@ -166,7 +168,9 @@ public: this->connections[remoteNodeId] = std::make_shared(ProxyConnection{ - .direction = request.direction(), + .direction = request.has_direction() + ? std::optional(request.direction()) + : std::nullopt, .userId = toEthereumAddress( BinaryUtils::binaryStringToHex(request.userid(), true)), .remote = std::move(ContentDeliveryRpcRemote( diff --git a/packages/streamr-trackerless-network/modules/logic/proxy/ProxyConnectionRpcRemote.cppm b/packages/streamr-trackerless-network/modules/logic/proxy/ProxyConnectionRpcRemote.cppm index 6d1f4d07..0be62db7 100644 --- a/packages/streamr-trackerless-network/modules/logic/proxy/ProxyConnectionRpcRemote.cppm +++ b/packages/streamr-trackerless-network/modules/logic/proxy/ProxyConnectionRpcRemote.cppm @@ -51,9 +51,13 @@ public: client, timeout) {} folly::coro::Task requestConnection( - ProxyDirection direction, const EthereumAddress& userId) { + std::optional direction, + const EthereumAddress& userId) { ProxyConnectionRequest request; - request.set_direction(direction); + // Absent direction = bidirectional (TS sends undefined). + if (direction.has_value()) { + request.set_direction(direction.value()); + } request.set_userid(BinaryUtils::hexToBinaryString(userId)); auto options = this->formDhtRpcOptions(); diff --git a/packages/streamr-trackerless-network/test/unit/ExternalNetworkRpcTest.cpp b/packages/streamr-trackerless-network/test/unit/ExternalNetworkRpcTest.cpp new file mode 100644 index 00000000..2d82ef88 --- /dev/null +++ b/packages/streamr-trackerless-network/test/unit/ExternalNetworkRpcTest.cpp @@ -0,0 +1,111 @@ +// Ported from packages/trackerless-network/test/end-to-end/ +// external-network-rpc.test.ts (v103.8.0-rc.3): an application registers +// a custom RPC method on the server node's external-network service and +// a client node queries it over a real websocket connection. +// +// NB: TestUtils and the textual pb.h are avoided — this TU composes the +// full NetworkNode + NetworkStack + DhtNode module graph and additional +// BMIs exhaust clang's per-TU source-location space (see the C3/C5 test +// files). +#include +#include +#include + +#include // IWYU pragma: keep + +import streamr.utils.CoroutineHelper; +import streamr.trackerlessnetwork.NetworkNode; +import streamr.trackerlessnetwork.NetworkStack; +import streamr.trackerlessnetwork.NetworkRpcClient; +import streamr.trackerlessnetwork.protos; +import streamr.dht.DhtCallContext; +import streamr.dht.DhtNode; +import streamr.dht.Identifiers; +import streamr.dht.protos; + +using ::dht::PeerDescriptor; +using streamr::dht::DhtNodeOptions; +using streamr::dht::Identifiers; +using streamr::dht::rpcprotocol::DhtCallContext; +using streamr::trackerlessnetwork::NetworkNode; +using streamr::trackerlessnetwork::NetworkOptions; +using streamr::trackerlessnetwork::NetworkStack; +using streamr::utils::blockingWait; +using HandshakeRpcClient = + streamr::protorpc::HandshakeRpcClient; + +namespace { + +constexpr uint16_t serverPort = 15499; + +inline PeerDescriptor createMockPeerDescriptor() { + PeerDescriptor descriptor; + descriptor.set_nodeid( + Identifiers::getRawFromDhtAddress( + Identifiers::createRandomDhtAddress())); + descriptor.set_type(::dht::NodeType::NODEJS); + return descriptor; +} + +} // namespace + +class ExternalNetworkRpcTest : public ::testing::Test { +protected: + PeerDescriptor serverPeerDescriptor; + PeerDescriptor clientPeerDescriptor = createMockPeerDescriptor(); + std::shared_ptr clientNode; + std::shared_ptr serverNode; + + void SetUp() override { + this->serverPeerDescriptor = createMockPeerDescriptor(); + this->serverPeerDescriptor.mutable_websocket()->set_host("127.0.0.1"); + this->serverPeerDescriptor.mutable_websocket()->set_port(serverPort); + this->serverPeerDescriptor.mutable_websocket()->set_tls(false); + + this->clientNode = std::make_shared( + std::make_shared(NetworkOptions{ + .layer0 = DhtNodeOptions{ + .peerDescriptor = this->clientPeerDescriptor, + .entryPoints = {this->serverPeerDescriptor}}})); + this->serverNode = std::make_shared( + std::make_shared(NetworkOptions{ + .layer0 = DhtNodeOptions{ + .peerDescriptor = this->serverPeerDescriptor, + .entryPoints = {this->serverPeerDescriptor}}})); + + blockingWait(this->serverNode->start()); + blockingWait(this->clientNode->start()); + } + + void TearDown() override { + if (this->serverNode) { + blockingWait(this->serverNode->stop()); + } + if (this->clientNode) { + blockingWait(this->clientNode->stop()); + } + } +}; + +TEST_F(ExternalNetworkRpcTest, CanMakeQueries) { + const std::string requestId = "TEST"; + this->serverNode->registerExternalNetworkRpcMethod< + StreamPartHandshakeRequest, + StreamPartHandshakeResponse>( + "handshake", + [&requestId]( + const StreamPartHandshakeRequest& /* request */, + const DhtCallContext& /* context */) { + StreamPartHandshakeResponse response; + response.set_requestid(requestId); + return response; + }); + auto client = + this->clientNode->createExternalRpcClient(); + DhtCallContext context; + context.sourceDescriptor = this->clientPeerDescriptor; + context.targetDescriptor = this->serverPeerDescriptor; + const auto response = blockingWait(client.handshake( + StreamPartHandshakeRequest{}, std::move(context), std::nullopt)); + EXPECT_EQ(response.requestid(), requestId); +} diff --git a/packages/streamr-trackerless-network/test/unit/NetworkNodeIntegrationTest.cpp b/packages/streamr-trackerless-network/test/unit/NetworkNodeIntegrationTest.cpp new file mode 100644 index 00000000..84071273 --- /dev/null +++ b/packages/streamr-trackerless-network/test/unit/NetworkNodeIntegrationTest.cpp @@ -0,0 +1,165 @@ +// Ported from packages/trackerless-network/test/integration/ +// NetworkNode.test.ts (v103.8.0-rc.3): two NetworkNodes over layer-0 +// simulator transports — join + broadcast/subscribe and fetchNodeInfo. +// +// NB: TestUtils and the textual pb.h are avoided — this TU composes the +// full NetworkStack + DhtNode + simulator module graph and additional +// BMIs exhaust clang's per-TU source-location space (see the C3/C5 test +// files). +#include +#include +#include +#include + +#include // IWYU pragma: keep + +import streamr.utils.CoroutineHelper; +import streamr.trackerlessnetwork.NetworkNode; +import streamr.trackerlessnetwork.NetworkStack; +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::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::createNetworkNode; +using streamr::trackerlessnetwork::NetworkNode; +using streamr::trackerlessnetwork::NetworkOptions; +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 (see the NB above). +inline PeerDescriptor createMockPeerDescriptor() { + PeerDescriptor descriptor; + descriptor.set_nodeid( + Identifiers::getRawFromDhtAddress( + Identifiers::createRandomDhtAddress())); + descriptor.set_type(::dht::NodeType::NODEJS); + return descriptor; +} + +StreamMessage createTestMessage(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_timestamp(666); // NOLINT + messageId->set_sequencenumber(0); + messageId->set_publisherid( + BinaryUtils::hexToBinaryString( + "0x1234567890123456789012345678901234567890")); + messageId->set_messagechainid("msgChainId"); + auto* previousMessageRef = msg.mutable_previousmessageref(); + previousMessageRef->set_timestamp(665); // NOLINT + previousMessageRef->set_sequencenumber(0); + auto* contentMessage = msg.mutable_contentmessage(); + contentMessage->set_content(R"({ "hello": "world" })"); + contentMessage->set_contenttype(ContentType::JSON); + contentMessage->set_encryptiontype(EncryptionType::NONE); + msg.set_signaturetype(SignatureType::ECDSA_SECP256K1_EVM); + msg.set_signature(BinaryUtils::hexToBinaryString("0x1234")); + return msg; +} + +} // namespace + +class NetworkNodeIntegrationTest : public ::testing::Test { +protected: + StreamPartID streamPartId = StreamPartIDUtils::parse("test#0"); + PeerDescriptor pd1 = createMockPeerDescriptor(); + PeerDescriptor pd2 = createMockPeerDescriptor(); + Simulator simulator{LatencyType::NONE}; + std::shared_ptr transport1; + std::shared_ptr transport2; + std::shared_ptr node1; + std::shared_ptr node2; + + void SetUp() override { + this->transport1 = + std::make_shared(this->pd1, this->simulator); + this->transport1->start(); + this->transport2 = + std::make_shared(this->pd2, this->simulator); + this->transport2->start(); + + this->node1 = createNetworkNode( + NetworkOptions{ + .layer0 = DhtNodeOptions{ + .transport = this->transport1.get(), + .connectionsView = this->transport1.get(), + .peerDescriptor = this->pd1, + .entryPoints = {this->pd1}}}); + this->node2 = createNetworkNode( + NetworkOptions{ + .layer0 = DhtNodeOptions{ + .transport = this->transport2.get(), + .connectionsView = this->transport2.get(), + .peerDescriptor = this->pd2, + .entryPoints = {this->pd1}}}); + + blockingWait(this->node1->start()); + this->node1->setStreamPartEntryPoints(this->streamPartId, {this->pd1}); + blockingWait(this->node2->start()); + this->node2->setStreamPartEntryPoints(this->streamPartId, {this->pd1}); + } + + void TearDown() override { + if (this->node1) { + blockingWait(this->node1->stop()); + } + if (this->node2) { + blockingWait(this->node2->stop()); + } + this->transport1->stop(); + this->transport2->stop(); + this->simulator.stop(); + } +}; + +TEST_F(NetworkNodeIntegrationTest, WaitForJoinPlusBroadcastAndSubscribe) { + std::atomic msgCount{0}; + blockingWait(this->node1->join(this->streamPartId)); + this->node1->addMessageListener([&msgCount](const StreamMessage& msg) { + EXPECT_EQ(msg.messageid().timestamp(), 666); + EXPECT_EQ(msg.messageid().sequencenumber(), 0); + msgCount += 1; + }); + blockingWait(this->node2->broadcast(createTestMessage(this->streamPartId))); + blockingWait( + waitForCondition([&msgCount]() { return msgCount.load() == 1; })); +} + +TEST_F(NetworkNodeIntegrationTest, FetchNodeInfo) { + blockingWait(this->node1->join(this->streamPartId)); + blockingWait(this->node2->join(this->streamPartId)); + const auto result1 = blockingWait(this->node1->fetchNodeInfo(this->pd2)); + const auto result2 = blockingWait(this->node2->fetchNodeInfo(this->pd1)); + const auto result3 = blockingWait( + this->node1->fetchNodeInfo(this->node1->getPeerDescriptor())); + EXPECT_EQ(result1.streampartitions_size(), 1); + EXPECT_EQ(result2.streampartitions_size(), 1); + EXPECT_EQ(result3.streampartitions_size(), 1); + EXPECT_EQ(result1.controllayer().connections_size(), 1); + EXPECT_EQ(result2.controllayer().connections_size(), 1); + EXPECT_EQ(result3.controllayer().connections_size(), 1); + EXPECT_EQ(result1.controllayer().neighbors_size(), 1); + EXPECT_EQ(result2.controllayer().neighbors_size(), 1); + EXPECT_EQ(result3.controllayer().neighbors_size(), 1); +} diff --git a/packages/streamr-trackerless-network/test/unit/NetworkNodeTest.cpp b/packages/streamr-trackerless-network/test/unit/NetworkNodeTest.cpp new file mode 100644 index 00000000..a428b75a --- /dev/null +++ b/packages/streamr-trackerless-network/test/unit/NetworkNodeTest.cpp @@ -0,0 +1,89 @@ +// Ported from packages/trackerless-network/test/unit/NetworkNode.test.ts +// (v103.8.0-rc.3): the message listener add/remove behaviour over a +// stubbed stack. The TS test injects a partial NetworkStack object; here +// a subclass stubs joinStreamPart (the stack is never started, so the +// real ContentDeliveryManager acts as the bare event emitter the TS +// test builds from EventEmitter). +#include +#include +#include +#include + +#include // IWYU pragma: keep + +import streamr.utils.CoroutineHelper; +import streamr.trackerlessnetwork.ContentDeliveryManager; +import streamr.trackerlessnetwork.NetworkNode; +import streamr.trackerlessnetwork.NetworkStack; +import streamr.trackerlessnetwork.protos; +import streamr.utils.BinaryUtils; +import streamr.utils.StreamPartID; + +using streamr::trackerlessnetwork::NeighborRequirement; +using streamr::trackerlessnetwork::NetworkNode; +using streamr::trackerlessnetwork::NetworkOptions; +using streamr::trackerlessnetwork::NetworkStack; +using streamr::trackerlessnetwork::contentdeliverymanagerevents::NewMessage; +using streamr::utils::BinaryUtils; +using streamr::utils::blockingWait; +using streamr::utils::StreamPartID; +using streamr::utils::StreamPartIDUtils; + +namespace { + +StreamMessage createMessage(int64_t id, const StreamPartID& streamPart) { + StreamMessage msg; + auto* messageId = msg.mutable_messageid(); + messageId->set_streamid(StreamPartIDUtils::getStreamID(streamPart)); + messageId->set_streampartition(0); + messageId->set_sequencenumber(0); + messageId->set_timestamp(id); + 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(std::to_string(id)); + return msg; +} + +class StubNetworkStack : public NetworkStack { +public: + using NetworkStack::NetworkStack; + + folly::coro::Task joinStreamPart( + StreamPartID /* streamPartId */, + std::optional /* neighborRequirement */) override { + co_return; + } +}; + +} // namespace + +TEST(NetworkNodeTest, MessageListener) { + const StreamPartID streamPart = StreamPartIDUtils::parse("stream#0"); + auto stack = std::make_shared(NetworkOptions{}); + NetworkNode node(stack); + blockingWait(node.join(streamPart)); + + std::vector received; + const auto token = node.addMessageListener( + [&received](const StreamMessage& msg) { received.push_back(msg); }); + + const auto msg1 = createMessage(1, streamPart); + const auto msg2 = createMessage(2, streamPart); + auto& manager = stack->getContentDeliveryManager(); + manager.emit(msg1); + manager.emit(msg2); + ASSERT_EQ(received.size(), 2); + EXPECT_EQ(received[0].messageid().timestamp(), 1); + EXPECT_EQ(received[1].messageid().timestamp(), 2); + + node.removeMessageListener(token); + manager.emit(createMessage(3, streamPart)); + EXPECT_EQ(received.size(), 2); +} diff --git a/packages/streamr-trackerless-network/test/unit/NetworkRpcTest.cpp b/packages/streamr-trackerless-network/test/unit/NetworkRpcTest.cpp new file mode 100644 index 00000000..6f41a1d8 --- /dev/null +++ b/packages/streamr-trackerless-network/test/unit/NetworkRpcTest.cpp @@ -0,0 +1,86 @@ +// Ported from packages/trackerless-network/test/integration/ +// NetworkRpc.test.ts (v103.8.0-rc.3): two raw RpcCommunicators wired +// back-to-back; the ContentDeliveryRpc client sends a stream message as +// a notification. +#include +#include +#include + +#include // IWYU pragma: keep + +import streamr.utils.CoroutineHelper; +import streamr.protorpc.RpcCommunicator; +import streamr.trackerlessnetwork.NetworkRpcClient; +import streamr.trackerlessnetwork.protos; +import streamr.dht.DhtCallContext; +import streamr.utils.BinaryUtils; +import streamr.utils.StreamPartID; +import streamr.utils.waitForCondition; + +using streamr::dht::rpcprotocol::DhtCallContext; +using streamr::protorpc::RpcCommunicator; +using streamr::protorpc::RpcMessage; +using streamr::utils::BinaryUtils; +using streamr::utils::blockingWait; +using streamr::utils::StreamPartID; +using streamr::utils::StreamPartIDUtils; +using streamr::utils::waitForCondition; +using ContentDeliveryRpcClient = + streamr::protorpc::ContentDeliveryRpcClient; + +namespace { + +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 + +TEST(NetworkRpcTest, SendsData) { + RpcCommunicator rpcCommunicator1; + RpcCommunicator rpcCommunicator2; + std::atomic recvCounter{0}; + + rpcCommunicator1.setOutgoingMessageCallback( + [&rpcCommunicator2]( + const RpcMessage& message, + const std::string& /* requestId */, + const DhtCallContext& /* context */) { + rpcCommunicator2.handleIncomingMessage(message, DhtCallContext()); + }); + rpcCommunicator2.registerRpcNotification( + "sendStreamMessage", + [&recvCounter]( + const StreamMessage& /* msg */, + const DhtCallContext& /* context */) { recvCounter += 1; }); + + ContentDeliveryRpcClient client{rpcCommunicator1}; + const auto msg = createLocalStreamMessage( + R"({ "hello": "WORLD" })", StreamPartIDUtils::parse("testStream#0")); + blockingWait(client.sendStreamMessage( + StreamMessage(msg), DhtCallContext(), std::nullopt)); + blockingWait( + waitForCondition([&recvCounter]() { return recvCounter.load() == 1; })); + + rpcCommunicator1.drainAsyncTasks(); + rpcCommunicator2.drainAsyncTasks(); +} diff --git a/packages/streamr-trackerless-network/test/unit/NetworkStackTest.cpp b/packages/streamr-trackerless-network/test/unit/NetworkStackTest.cpp new file mode 100644 index 00000000..a00888f9 --- /dev/null +++ b/packages/streamr-trackerless-network/test/unit/NetworkStackTest.cpp @@ -0,0 +1,161 @@ +// Ported from packages/trackerless-network/test/integration/ +// NetworkStack.test.ts (v103.8.0-rc.3): two stacks over REAL websockets +// (no transport given — each layer-0 node owns a ConnectionManager) — +// pub/sub through the content delivery manager and joinStreamPart with +// a neighbor requirement. +// +// NB: TestUtils and the textual pb.h are avoided — this TU composes the +// full NetworkStack + DhtNode module graph and additional BMIs exhaust +// clang's per-TU source-location space (see the C3/C5 test files). +#include +#include +#include +#include +#include + +#include // IWYU pragma: keep + +import streamr.utils.CoroutineHelper; +import streamr.trackerlessnetwork.ContentDeliveryManager; +import streamr.trackerlessnetwork.NetworkStack; +import streamr.trackerlessnetwork.protos; +import streamr.dht.DhtNode; +import streamr.dht.Identifiers; +import streamr.dht.PortRange; +import streamr.dht.protos; +import streamr.utils.BinaryUtils; +import streamr.utils.StreamPartID; +import streamr.utils.waitForCondition; + +using ::dht::PeerDescriptor; +using streamr::dht::DhtNodeOptions; +using streamr::dht::Identifiers; +using streamr::dht::types::PortRange; +using streamr::trackerlessnetwork::NeighborRequirement; +using streamr::trackerlessnetwork::NetworkOptions; +using streamr::trackerlessnetwork::NetworkStack; +using streamr::trackerlessnetwork::contentdeliverymanagerevents::NewMessage; +using streamr::utils::BinaryUtils; +using streamr::utils::blockingWait; +using streamr::utils::StreamPartID; +using streamr::utils::StreamPartIDUtils; +using streamr::utils::waitForCondition; + +namespace { + +constexpr uint16_t entryPointPort = 32222; +constexpr uint16_t otherPort = 32223; + +inline PeerDescriptor createMockPeerDescriptor() { + PeerDescriptor descriptor; + descriptor.set_nodeid( + Identifiers::getRawFromDhtAddress( + Identifiers::createRandomDhtAddress())); + descriptor.set_type(::dht::NodeType::NODEJS); + return descriptor; +} + +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 NetworkStackTest : public ::testing::Test { +protected: + StreamPartID streamPartId = StreamPartIDUtils::parse("stream#0"); + PeerDescriptor epDescriptor; + std::shared_ptr stack1; + std::shared_ptr stack2; + + void SetUp() override { + this->epDescriptor = createMockPeerDescriptor(); + this->epDescriptor.mutable_websocket()->set_host("127.0.0.1"); + this->epDescriptor.mutable_websocket()->set_port(entryPointPort); + this->epDescriptor.mutable_websocket()->set_tls(false); + + this->stack1 = std::make_shared(NetworkOptions{ + .layer0 = DhtNodeOptions{ + .peerDescriptor = this->epDescriptor, + .entryPoints = {this->epDescriptor}}}); + this->stack2 = std::make_shared(NetworkOptions{ + .layer0 = DhtNodeOptions{ + .entryPoints = {this->epDescriptor}, + .websocketPortRange = + PortRange{.min = otherPort, .max = otherPort}}}); + + blockingWait(this->stack1->start()); + this->stack1->getContentDeliveryManager().setStreamPartEntryPoints( + this->streamPartId, {this->epDescriptor}); + blockingWait(this->stack2->start()); + this->stack2->getContentDeliveryManager().setStreamPartEntryPoints( + this->streamPartId, {this->epDescriptor}); + } + + void TearDown() override { + if (this->stack1) { + blockingWait(this->stack1->stop()); + } + if (this->stack2) { + blockingWait(this->stack2->stop()); + } + } +}; + +TEST_F(NetworkStackTest, CanUseNetworkNodePubSubViaNetworkStack) { + std::atomic receivedMessages{0}; + this->stack1->getContentDeliveryManager().joinStreamPart( + this->streamPartId); + this->stack1->getContentDeliveryManager().on( + [&receivedMessages](const StreamMessage& /* msg */) { + receivedMessages += 1; + }); + const auto msg = + createLocalStreamMessage(R"({ "hello": "WORLD" })", this->streamPartId); + this->stack2->getContentDeliveryManager().broadcast(msg); + blockingWait(waitForCondition( + [&receivedMessages]() { return receivedMessages.load() == 1; }, + std::chrono::seconds(15))); // NOLINT +} + +TEST_F(NetworkStackTest, JoinAndWaitForNeighbors) { + constexpr auto neighborTimeout = std::chrono::seconds(5); + blockingWait( + folly::coro::collectAll( + this->stack1->joinStreamPart( + this->streamPartId, + NeighborRequirement{.minCount = 1, .timeout = neighborTimeout}), + this->stack2->joinStreamPart( + this->streamPartId, + NeighborRequirement{ + .minCount = 1, .timeout = neighborTimeout}))); + EXPECT_EQ( + this->stack1->getContentDeliveryManager() + .getNeighbors(this->streamPartId) + .size(), + 1); + EXPECT_EQ( + this->stack2->getContentDeliveryManager() + .getNeighbors(this->streamPartId) + .size(), + 1); +} diff --git a/packages/streamr-trackerless-network/test/unit/NodeInfoRpcTest.cpp b/packages/streamr-trackerless-network/test/unit/NodeInfoRpcTest.cpp new file mode 100644 index 00000000..bfe7dfe3 --- /dev/null +++ b/packages/streamr-trackerless-network/test/unit/NodeInfoRpcTest.cpp @@ -0,0 +1,185 @@ +// Ported from packages/trackerless-network/test/integration/ +// NodeInfoRpc.test.ts (v103.8.0-rc.3): a NodeInfoClient on a third +// transport queries a NetworkStack that shares two stream parts with +// another stack. +// +// NB: TestUtils and the textual pb.h are avoided — this TU composes the +// full NetworkStack + DhtNode + simulator module graph and additional +// BMIs exhaust clang's per-TU source-location space (see the C3/C5 test +// files). +#include +#include +#include +#include + +#include // IWYU pragma: keep + +import streamr.utils.CoroutineHelper; +import streamr.trackerlessnetwork.NetworkStack; +import streamr.trackerlessnetwork.NodeInfoClient; +import streamr.trackerlessnetwork.NodeInfoRpcLocal; +import streamr.trackerlessnetwork.protos; +import streamr.dht.DhtNode; +import streamr.dht.Identifiers; +import streamr.dht.ListeningRpcCommunicator; +import streamr.dht.Simulator; +import streamr.dht.SimulatorTransport; +import streamr.dht.protos; +import streamr.utils.StreamPartID; +import streamr.utils.waitForCondition; + +using ::dht::PeerDescriptor; +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::dht::transport::ListeningRpcCommunicator; +using streamr::trackerlessnetwork::NetworkOptions; +using streamr::trackerlessnetwork::NetworkStack; +using streamr::trackerlessnetwork::NodeInfoClient; +using streamr::trackerlessnetwork::nodeInfoRpcServiceId; +using streamr::utils::blockingWait; +using streamr::utils::StreamPartID; +using streamr::utils::StreamPartIDUtils; +using streamr::utils::waitForCondition; + +namespace { + +inline PeerDescriptor createMockPeerDescriptor() { + PeerDescriptor descriptor; + descriptor.set_nodeid( + Identifiers::getRawFromDhtAddress( + Identifiers::createRandomDhtAddress())); + descriptor.set_type(::dht::NodeType::NODEJS); + return descriptor; +} + +bool containsPeer(const auto& repeatedField, const PeerDescriptor& expected) { + return std::ranges::any_of( + repeatedField, [&expected](const PeerDescriptor& descriptor) { + return descriptor.nodeid() == expected.nodeid(); + }); +} + +} // namespace + +class NodeInfoRpcTest : public ::testing::Test { +protected: + PeerDescriptor requesteePeerDescriptor = createMockPeerDescriptor(); + PeerDescriptor otherPeerDescriptor = createMockPeerDescriptor(); + PeerDescriptor requestorPeerDescriptor = createMockPeerDescriptor(); + Simulator simulator{LatencyType::NONE}; + std::shared_ptr requesteeTransport; + std::shared_ptr otherTransport; + std::shared_ptr requestorTransport; + std::shared_ptr requesteeStack; + std::shared_ptr otherStack; + std::unique_ptr requestorCommunicator; + std::unique_ptr nodeInfoClient; + + void SetUp() override { + this->requesteeTransport = std::make_shared( + this->requesteePeerDescriptor, this->simulator); + this->otherTransport = std::make_shared( + this->otherPeerDescriptor, this->simulator); + this->requestorTransport = std::make_shared( + this->requestorPeerDescriptor, this->simulator); + this->requesteeTransport->start(); + this->otherTransport->start(); + this->requestorTransport->start(); + this->requesteeStack = std::make_shared(NetworkOptions{ + .layer0 = DhtNodeOptions{ + .transport = this->requesteeTransport.get(), + .connectionsView = this->requesteeTransport.get(), + .peerDescriptor = this->requesteePeerDescriptor, + .entryPoints = {this->requesteePeerDescriptor}}}); + this->otherStack = std::make_shared(NetworkOptions{ + .layer0 = DhtNodeOptions{ + .transport = this->otherTransport.get(), + .connectionsView = this->otherTransport.get(), + .peerDescriptor = this->otherPeerDescriptor, + .entryPoints = {this->requesteePeerDescriptor}}}); + blockingWait(this->requesteeStack->start()); + blockingWait(this->otherStack->start()); + this->requestorCommunicator = + std::make_unique( + ServiceID{nodeInfoRpcServiceId}, *this->requestorTransport); + this->nodeInfoClient = std::make_unique( + this->requestorPeerDescriptor, *this->requestorCommunicator); + } + + void TearDown() override { + if (this->requesteeStack) { + blockingWait(this->requesteeStack->stop()); + } + if (this->otherStack) { + blockingWait(this->otherStack->stop()); + } + if (this->requestorCommunicator) { + this->requestorCommunicator->destroy(); + } + this->requesteeTransport->stop(); + this->otherTransport->stop(); + this->requestorTransport->stop(); + this->simulator.stop(); + } +}; + +TEST_F(NodeInfoRpcTest, HappyPath) { + const auto streamPartId1 = StreamPartIDUtils::parse("stream1#0"); + const auto streamPartId2 = StreamPartIDUtils::parse("stream2#0"); + this->requesteeStack->getContentDeliveryManager().joinStreamPart( + streamPartId1); + this->otherStack->getContentDeliveryManager().joinStreamPart(streamPartId1); + this->requesteeStack->getContentDeliveryManager().joinStreamPart( + streamPartId2); + this->otherStack->getContentDeliveryManager().joinStreamPart(streamPartId2); + blockingWait(waitForCondition( + [this, &streamPartId1, &streamPartId2]() { + return this->requesteeStack->getContentDeliveryManager() + .getNeighbors(streamPartId1) + .size() == 1 && + this->otherStack->getContentDeliveryManager() + .getNeighbors(streamPartId1) + .size() == 1 && + this->requesteeStack->getContentDeliveryManager() + .getNeighbors(streamPartId2) + .size() == 1 && + this->otherStack->getContentDeliveryManager() + .getNeighbors(streamPartId2) + .size() == 1; + }, + std::chrono::seconds(15))); // NOLINT + + const auto result = blockingWait( + this->nodeInfoClient->getInfo(this->requesteePeerDescriptor)); + + EXPECT_EQ( + result.peerdescriptor().nodeid(), + this->requesteePeerDescriptor.nodeid()); + ASSERT_EQ(result.controllayer().neighbors_size(), 1); + EXPECT_TRUE(containsPeer( + result.controllayer().neighbors(), this->otherPeerDescriptor)); + EXPECT_EQ(result.controllayer().connections_size(), 2); + EXPECT_TRUE(containsPeer( + result.controllayer().connections(), this->otherPeerDescriptor)); + EXPECT_TRUE(containsPeer( + result.controllayer().connections(), this->requestorPeerDescriptor)); + ASSERT_EQ(result.streampartitions_size(), 2); + for (const auto& partition : result.streampartitions()) { + EXPECT_TRUE( + partition.id() == streamPartId1 || partition.id() == streamPartId2); + ASSERT_EQ(partition.controllayerneighbors_size(), 1); + EXPECT_TRUE(containsPeer( + partition.controllayerneighbors(), this->otherPeerDescriptor)); + ASSERT_EQ(partition.contentdeliverylayerneighbors_size(), 1); + EXPECT_EQ( + partition.contentdeliverylayerneighbors(0) + .peerdescriptor() + .nodeid(), + this->otherPeerDescriptor.nodeid()); + } + EXPECT_FALSE(result.applicationversion().empty()); +} diff --git a/packages/streamr-trackerless-network/test/unit/ProxyAndFullNodeTest.cpp b/packages/streamr-trackerless-network/test/unit/ProxyAndFullNodeTest.cpp new file mode 100644 index 00000000..541b317c --- /dev/null +++ b/packages/streamr-trackerless-network/test/unit/ProxyAndFullNodeTest.cpp @@ -0,0 +1,225 @@ +// Ported from packages/trackerless-network/test/end-to-end/ +// proxy-and-full-node.test.ts (v103.8.0-rc.3, deferred from phase C5): +// a node with only proxy connections on one stream part still acts as a +// FULL node on other stream parts (joining the control layer on first +// non-proxied broadcast). +// +// NB: TestUtils and the textual pb.h are avoided — this TU composes the +// full NetworkNode + NetworkStack + DhtNode module graph and additional +// BMIs exhaust clang's per-TU source-location space (see the C3/C5 test +// files). +#include +#include +#include +#include +#include +#include +#include + +#include // IWYU pragma: keep + +import streamr.utils.CoroutineHelper; +import streamr.trackerlessnetwork.ContentDeliveryManager; +import streamr.trackerlessnetwork.NetworkNode; +import streamr.trackerlessnetwork.NetworkStack; +import streamr.trackerlessnetwork.protos; +import streamr.dht.DhtNode; +import streamr.dht.Identifiers; +import streamr.dht.protos; +import streamr.utils.BinaryUtils; +import streamr.utils.EthereumAddress; +import streamr.utils.StreamPartID; +import streamr.utils.waitForCondition; + +using ::dht::PeerDescriptor; +using streamr::dht::DhtNodeOptions; +using streamr::dht::Identifiers; +using streamr::trackerlessnetwork::ContentDeliveryManagerOptions; +using streamr::trackerlessnetwork::createNetworkNode; +using streamr::trackerlessnetwork::NetworkNode; +using streamr::trackerlessnetwork::NetworkOptions; +using streamr::trackerlessnetwork::contentdeliverymanagerevents::NewMessage; +using streamr::utils::BinaryUtils; +using streamr::utils::blockingWait; +using streamr::utils::EthereumAddress; +using streamr::utils::StreamPartID; +using streamr::utils::StreamPartIDUtils; +using streamr::utils::toEthereumAddress; +using streamr::utils::waitForCondition; + +namespace { + +constexpr uint16_t proxyPort = 23135; +constexpr auto defaultTimeout = std::chrono::seconds(10); + +inline PeerDescriptor createMockPeerDescriptor() { + PeerDescriptor descriptor; + descriptor.set_nodeid( + Identifiers::getRawFromDhtAddress( + Identifiers::createRandomDhtAddress())); + descriptor.set_type(::dht::NodeType::NODEJS); + return descriptor; +} + +StreamMessage createTestMessage(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_timestamp(666); // NOLINT + messageId->set_sequencenumber(0); + messageId->set_publisherid( + BinaryUtils::hexToBinaryString( + "0x1234567890123456789012345678901234567890")); + messageId->set_messagechainid("msgChainId"); + auto* previousMessageRef = msg.mutable_previousmessageref(); + previousMessageRef->set_timestamp(665); // NOLINT + previousMessageRef->set_sequencenumber(0); + auto* contentMessage = msg.mutable_contentmessage(); + contentMessage->set_content(R"({ "hello": "world" })"); + contentMessage->set_contenttype(ContentType::JSON); + contentMessage->set_encryptiontype(EncryptionType::NONE); + msg.set_signaturetype(SignatureType::ECDSA_SECP256K1_EVM); + msg.set_signature(BinaryUtils::hexToBinaryString("0x1234")); + return msg; +} + +} // namespace + +class ProxyAndFullNodeTest : public ::testing::Test { +protected: + StreamPartID proxiedStreamPart = StreamPartIDUtils::parse("proxy-stream#0"); + StreamPartID regularStreamPart1 = + StreamPartIDUtils::parse("regular-stream1#0"); + StreamPartID regularStreamPart2 = + StreamPartIDUtils::parse("regular-stream2#0"); + StreamPartID regularStreamPart3 = + StreamPartIDUtils::parse("regular-stream3#0"); + StreamPartID regularStreamPart4 = + StreamPartIDUtils::parse("regular-stream4#0"); + EthereumAddress proxiedNodeUserId = + toEthereumAddress("0x3333333333333333333333333333333333333333"); + std::shared_ptr proxyNode; + std::shared_ptr proxiedNode; + + void SetUp() override { + auto proxyNodeDescriptor = createMockPeerDescriptor(); + proxyNodeDescriptor.mutable_websocket()->set_host("127.0.0.1"); + proxyNodeDescriptor.mutable_websocket()->set_port(proxyPort); + proxyNodeDescriptor.mutable_websocket()->set_tls(false); + const auto proxiedNodeDescriptor = createMockPeerDescriptor(); + + this->proxyNode = createNetworkNode( + NetworkOptions{ + .layer0 = + DhtNodeOptions{ + .peerDescriptor = proxyNodeDescriptor, + .entryPoints = {proxyNodeDescriptor}}, + .networkNode = ContentDeliveryManagerOptions{ + .acceptProxyConnections = true}}); + blockingWait(this->proxyNode->start()); + auto& proxyManager = + this->proxyNode->getStack().getContentDeliveryManager(); + proxyManager.joinStreamPart(proxiedStreamPart); + proxyManager.joinStreamPart(regularStreamPart1); + proxyManager.joinStreamPart(regularStreamPart2); + proxyManager.joinStreamPart(regularStreamPart3); + proxyManager.joinStreamPart(regularStreamPart4); + + this->proxiedNode = createNetworkNode( + NetworkOptions{ + .layer0 = DhtNodeOptions{ + .peerDescriptor = proxiedNodeDescriptor, + .entryPoints = {proxyNodeDescriptor}}}); + blockingWait(this->proxiedNode->start(false)); + } + + // Null-guarded: when SetUp() throws (e.g. a websocket port is + // still in TIME_WAIT), gtest still runs TearDown() with later + // members never assigned. + void TearDown() override { + if (this->proxyNode) { + blockingWait(this->proxyNode->stop()); + } + if (this->proxiedNode) { + blockingWait(this->proxiedNode->stop()); + } + } + + void broadcastAndWaitForMessage(const StreamPartID& streamPartId) { + std::atomic received{false}; + auto& proxyManager = + this->proxyNode->getStack().getContentDeliveryManager(); + const auto token = proxyManager.on( + [&received, &streamPartId](const StreamMessage& msg) { + if (msg.messageid().streamid() == + StreamPartIDUtils::getStreamID(streamPartId)) { + received = true; + } + }); + blockingWait( + this->proxiedNode->broadcast(createTestMessage(streamPartId))); + blockingWait(waitForCondition( + [&received]() { return received.load(); }, defaultTimeout)); + proxyManager.off(token); + } +}; + +TEST_F(ProxyAndFullNodeTest, ProxiedNodeCanActAsFullNodeOnAnotherStreamPart) { + blockingWait(this->proxiedNode->setProxies( + proxiedStreamPart, + {this->proxyNode->getPeerDescriptor()}, + std::nullopt, + this->proxiedNodeUserId)); + EXPECT_FALSE( + this->proxiedNode->getStack().getControlLayerNode().hasJoined()); + + this->broadcastAndWaitForMessage(regularStreamPart1); + + EXPECT_TRUE( + this->proxiedNode->getStack().getControlLayerNode().hasJoined()); + + this->broadcastAndWaitForMessage(proxiedStreamPart); + + auto& proxiedManager = + this->proxiedNode->getStack().getContentDeliveryManager(); + EXPECT_TRUE( + proxiedManager.getStreamPartDelivery(proxiedStreamPart)->proxied); + EXPECT_FALSE( + proxiedManager.getStreamPartDelivery(regularStreamPart1)->proxied); +} + +TEST_F(ProxyAndFullNodeTest, ProxiedNodeCanActAsFullNodeOnMultipleStreamParts) { + blockingWait(this->proxiedNode->setProxies( + proxiedStreamPart, + {this->proxyNode->getPeerDescriptor()}, + std::nullopt, + this->proxiedNodeUserId)); + EXPECT_FALSE( + this->proxiedNode->getStack().getControlLayerNode().hasJoined()); + + this->broadcastAndWaitForMessage(regularStreamPart1); + this->broadcastAndWaitForMessage(regularStreamPart2); + this->broadcastAndWaitForMessage(regularStreamPart3); + this->broadcastAndWaitForMessage(regularStreamPart4); + + EXPECT_TRUE( + this->proxiedNode->getStack().getControlLayerNode().hasJoined()); + + this->broadcastAndWaitForMessage(proxiedStreamPart); + + auto& proxiedManager = + this->proxiedNode->getStack().getContentDeliveryManager(); + EXPECT_TRUE( + proxiedManager.getStreamPartDelivery(proxiedStreamPart)->proxied); + EXPECT_FALSE( + proxiedManager.getStreamPartDelivery(regularStreamPart1)->proxied); + EXPECT_FALSE( + proxiedManager.getStreamPartDelivery(regularStreamPart2)->proxied); + EXPECT_FALSE( + proxiedManager.getStreamPartDelivery(regularStreamPart3)->proxied); + EXPECT_FALSE( + proxiedManager.getStreamPartDelivery(regularStreamPart4)->proxied); +} diff --git a/packages/streamr-trackerless-network/test/unit/ProxyConnectionsTest.cpp b/packages/streamr-trackerless-network/test/unit/ProxyConnectionsTest.cpp new file mode 100644 index 00000000..9f79ac3a --- /dev/null +++ b/packages/streamr-trackerless-network/test/unit/ProxyConnectionsTest.cpp @@ -0,0 +1,402 @@ +// Ported from packages/trackerless-network/test/end-to-end/ +// proxy-connections.test.ts (v103.8.0-rc.3, deferred from phase C5): +// two proxy-accepting full nodes over real websockets and one proxied +// node — publish/subscribe through proxies, opening/closing/leaving +// connections, bidirectional (direction-less) proxies, reconnect after +// the proxy leaves and rejoins, and the join/broadcast guards. +// +// NB: TestUtils and the textual pb.h are avoided — this TU composes the +// full NetworkNode + NetworkStack + DhtNode module graph and additional +// BMIs exhaust clang's per-TU source-location space (see the C3/C5 test +// files). +#include +#include +#include +#include +#include +#include +#include + +#include // IWYU pragma: keep + +import streamr.utils.CoroutineHelper; +import streamr.trackerlessnetwork.ContentDeliveryManager; +import streamr.trackerlessnetwork.NetworkNode; +import streamr.trackerlessnetwork.NetworkStack; +import streamr.trackerlessnetwork.protos; +import streamr.dht.DhtNode; +import streamr.dht.Identifiers; +import streamr.dht.protos; +import streamr.utils.BinaryUtils; +import streamr.utils.EthereumAddress; +import streamr.utils.StreamPartID; +import streamr.utils.waitForCondition; + +using ::dht::PeerDescriptor; +using streamr::dht::DhtAddress; +using streamr::dht::DhtNodeOptions; +using streamr::dht::Identifiers; +using streamr::trackerlessnetwork::ContentDeliveryManagerOptions; +using streamr::trackerlessnetwork::createNetworkNode; +using streamr::trackerlessnetwork::NetworkNode; +using streamr::trackerlessnetwork::NetworkOptions; +using streamr::trackerlessnetwork::contentdeliverymanagerevents::NewMessage; +using streamr::utils::BinaryUtils; +using streamr::utils::blockingWait; +using streamr::utils::EthereumAddress; +using streamr::utils::StreamPartID; +using streamr::utils::StreamPartIDUtils; +using streamr::utils::toEthereumAddress; +using streamr::utils::waitForCondition; + +namespace { + +constexpr uint16_t proxyPort1 = 23132; +constexpr uint16_t proxyPort2 = 23133; +constexpr auto defaultTimeout = std::chrono::seconds(10); +constexpr auto reconnectTimeout = std::chrono::seconds(25); + +inline PeerDescriptor createMockPeerDescriptor() { + PeerDescriptor descriptor; + descriptor.set_nodeid( + Identifiers::getRawFromDhtAddress( + Identifiers::createRandomDhtAddress())); + descriptor.set_type(::dht::NodeType::NODEJS); + return descriptor; +} + +StreamMessage createTestMessage(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_timestamp(666); // NOLINT + messageId->set_sequencenumber(0); + messageId->set_publisherid( + BinaryUtils::hexToBinaryString( + "0x1234567890123456789012345678901234567890")); + messageId->set_messagechainid("msgChainId"); + auto* previousMessageRef = msg.mutable_previousmessageref(); + previousMessageRef->set_timestamp(665); // NOLINT + previousMessageRef->set_sequencenumber(0); + auto* contentMessage = msg.mutable_contentmessage(); + contentMessage->set_content(R"({ "hello": "world" })"); + contentMessage->set_contenttype(ContentType::JSON); + contentMessage->set_encryptiontype(EncryptionType::NONE); + msg.set_signaturetype(SignatureType::ECDSA_SECP256K1_EVM); + msg.set_signature(BinaryUtils::hexToBinaryString("0x1234")); + return msg; +} + +} // namespace + +class ProxyConnectionsTest : public ::testing::Test { +protected: + StreamPartID streamPartId = StreamPartIDUtils::parse("proxy-test#0"); + EthereumAddress proxiedNodeUserId = + toEthereumAddress("0x2222222222222222222222222222222222222222"); + std::shared_ptr proxyNode1; + std::shared_ptr proxyNode2; + std::shared_ptr proxiedNode; + + [[nodiscard]] bool hasConnectionFromProxy(NetworkNode& proxyNode) const { + const auto delivery = proxyNode.getStack() + .getContentDeliveryManager() + .getStreamPartDelivery(this->streamPartId); + return delivery != nullptr && delivery->node != nullptr && + delivery->node->hasProxyConnection(this->proxiedNode->getNodeId()); + } + + [[nodiscard]] bool hasConnectionToProxy( + const DhtAddress& proxyNodeId, + std::optional direction = std::nullopt) const { + const auto delivery = this->proxiedNode->getStack() + .getContentDeliveryManager() + .getStreamPartDelivery(this->streamPartId); + return delivery != nullptr && delivery->client != nullptr && + delivery->client->hasConnection(proxyNodeId, direction); + } + + void SetUp() override { + auto proxyNodeDescriptor1 = createMockPeerDescriptor(); + proxyNodeDescriptor1.mutable_websocket()->set_host("127.0.0.1"); + proxyNodeDescriptor1.mutable_websocket()->set_port(proxyPort1); + proxyNodeDescriptor1.mutable_websocket()->set_tls(false); + auto proxyNodeDescriptor2 = createMockPeerDescriptor(); + proxyNodeDescriptor2.mutable_websocket()->set_host("127.0.0.1"); + proxyNodeDescriptor2.mutable_websocket()->set_port(proxyPort2); + proxyNodeDescriptor2.mutable_websocket()->set_tls(false); + const auto proxiedNodeDescriptor = createMockPeerDescriptor(); + + this->proxyNode1 = createNetworkNode( + NetworkOptions{ + .layer0 = + DhtNodeOptions{ + .peerDescriptor = proxyNodeDescriptor1, + .entryPoints = {proxyNodeDescriptor1}}, + .networkNode = ContentDeliveryManagerOptions{ + .acceptProxyConnections = true}}); + blockingWait(this->proxyNode1->start()); + this->proxyNode1->getStack().getContentDeliveryManager().joinStreamPart( + this->streamPartId); + + this->proxyNode2 = createNetworkNode( + NetworkOptions{ + .layer0 = + DhtNodeOptions{ + .peerDescriptor = proxyNodeDescriptor2, + .entryPoints = {proxyNodeDescriptor1}}, + .networkNode = ContentDeliveryManagerOptions{ + .acceptProxyConnections = true}}); + blockingWait(this->proxyNode2->start()); + this->proxyNode2->getStack().getContentDeliveryManager().joinStreamPart( + this->streamPartId); + + this->proxiedNode = createNetworkNode( + NetworkOptions{ + .layer0 = DhtNodeOptions{ + .peerDescriptor = proxiedNodeDescriptor, + .entryPoints = {this->proxyNode1->getPeerDescriptor()}}}); + blockingWait(this->proxiedNode->start(false)); + } + + // Null-guarded: when SetUp() throws (e.g. a websocket port is + // still in TIME_WAIT), gtest still runs TearDown() with later + // members never assigned. + void TearDown() override { + if (this->proxyNode1) { + blockingWait(this->proxyNode1->stop()); + } + if (this->proxyNode2) { + blockingWait(this->proxyNode2->stop()); + } + if (this->proxiedNode) { + blockingWait(this->proxiedNode->stop()); + } + } + + // NOLINTNEXTLINE(bugprone-easily-swappable-parameters) + void waitForMessage(NetworkNode& receiver, NetworkNode& sender) const { + std::atomic received{false}; + const auto token = + receiver.getStack().getContentDeliveryManager().on( + [&received](const StreamMessage& /* msg */) { + received = true; + }); + blockingWait(sender.broadcast(createTestMessage(this->streamPartId))); + blockingWait(waitForCondition( + [&received]() { return received.load(); }, defaultTimeout)); + receiver.getStack().getContentDeliveryManager().off(token); + } +}; + +TEST_F(ProxyConnectionsTest, HappyPathPublishing) { + blockingWait(this->proxiedNode->setProxies( + this->streamPartId, + {this->proxyNode1->getPeerDescriptor()}, + ProxyDirection::PUBLISH, + this->proxiedNodeUserId, + 1)); + this->waitForMessage(*this->proxyNode1, *this->proxiedNode); +} + +TEST_F(ProxyConnectionsTest, HappyPathSubscribing) { + blockingWait(this->proxiedNode->setProxies( + this->streamPartId, + {this->proxyNode1->getPeerDescriptor()}, + ProxyDirection::SUBSCRIBE, + this->proxiedNodeUserId, + 1)); + this->waitForMessage(*this->proxiedNode, *this->proxyNode1); +} + +TEST_F(ProxyConnectionsTest, CanLeaveProxyPublishConnection) { + blockingWait(this->proxiedNode->setProxies( + this->streamPartId, + {this->proxyNode1->getPeerDescriptor()}, + ProxyDirection::PUBLISH, + this->proxiedNodeUserId, + 1)); + EXPECT_TRUE(this->proxiedNode->hasStreamPart(this->streamPartId)); + EXPECT_TRUE(this->hasConnectionFromProxy(*this->proxyNode1)); + blockingWait(this->proxiedNode->setProxies( + this->streamPartId, + {}, + ProxyDirection::PUBLISH, + this->proxiedNodeUserId, + 0)); + EXPECT_FALSE(this->proxiedNode->hasStreamPart(this->streamPartId)); + blockingWait(waitForCondition( + [this]() { return !this->hasConnectionFromProxy(*this->proxyNode1); }, + defaultTimeout)); +} + +TEST_F(ProxyConnectionsTest, CanLeaveProxySubscribeConnection) { + blockingWait(this->proxiedNode->setProxies( + this->streamPartId, + {this->proxyNode1->getPeerDescriptor()}, + ProxyDirection::SUBSCRIBE, + this->proxiedNodeUserId, + 1)); + EXPECT_TRUE(this->proxiedNode->hasStreamPart(this->streamPartId)); + EXPECT_TRUE(this->hasConnectionFromProxy(*this->proxyNode1)); + blockingWait(this->proxiedNode->setProxies( + this->streamPartId, + {}, + ProxyDirection::SUBSCRIBE, + this->proxiedNodeUserId, + 0)); + EXPECT_FALSE(this->proxiedNode->hasStreamPart(this->streamPartId)); + blockingWait(waitForCondition( + [this]() { return !this->hasConnectionFromProxy(*this->proxyNode1); }, + defaultTimeout)); +} + +TEST_F(ProxyConnectionsTest, CanOpenMultipleProxyConnections) { + blockingWait(this->proxiedNode->setProxies( + this->streamPartId, + {this->proxyNode1->getPeerDescriptor(), + this->proxyNode2->getPeerDescriptor()}, + std::nullopt, + this->proxiedNodeUserId)); + EXPECT_TRUE(this->proxiedNode->hasStreamPart(this->streamPartId)); + EXPECT_TRUE(this->hasConnectionFromProxy(*this->proxyNode1)); + EXPECT_TRUE(this->hasConnectionFromProxy(*this->proxyNode2)); +} + +TEST_F(ProxyConnectionsTest, CanOpenMultipleProxyConnectionsAndCloseOne) { + blockingWait(this->proxiedNode->setProxies( + this->streamPartId, + {this->proxyNode1->getPeerDescriptor(), + this->proxyNode2->getPeerDescriptor()}, + std::nullopt, + this->proxiedNodeUserId)); + EXPECT_TRUE(this->proxiedNode->hasStreamPart(this->streamPartId)); + EXPECT_TRUE(this->hasConnectionFromProxy(*this->proxyNode1)); + EXPECT_TRUE(this->hasConnectionFromProxy(*this->proxyNode2)); + blockingWait(this->proxiedNode->setProxies( + this->streamPartId, + {this->proxyNode1->getPeerDescriptor()}, + std::nullopt, + this->proxiedNodeUserId)); + EXPECT_TRUE(this->proxiedNode->hasStreamPart(this->streamPartId)); + blockingWait(waitForCondition( + [this]() { return !this->hasConnectionFromProxy(*this->proxyNode2); }, + defaultTimeout)); + EXPECT_TRUE(this->hasConnectionFromProxy(*this->proxyNode1)); +} + +TEST_F(ProxyConnectionsTest, CanOpenAndCloseAllConnections) { + blockingWait(this->proxiedNode->setProxies( + this->streamPartId, + {this->proxyNode1->getPeerDescriptor(), + this->proxyNode2->getPeerDescriptor()}, + std::nullopt, + this->proxiedNodeUserId)); + EXPECT_TRUE(this->proxiedNode->hasStreamPart(this->streamPartId)); + EXPECT_TRUE(this->hasConnectionFromProxy(*this->proxyNode1)); + EXPECT_TRUE(this->hasConnectionFromProxy(*this->proxyNode2)); + blockingWait(this->proxiedNode->setProxies( + this->streamPartId, {}, std::nullopt, this->proxiedNodeUserId)); + EXPECT_FALSE(this->proxiedNode->hasStreamPart(this->streamPartId)); + blockingWait(waitForCondition( + [this]() { return !this->hasConnectionFromProxy(*this->proxyNode1); }, + defaultTimeout)); + blockingWait(waitForCondition( + [this]() { return !this->hasConnectionFromProxy(*this->proxyNode2); }, + defaultTimeout)); +} + +// Adaptation of the TS test: upstream, the first `until` passes only +// because the JS event loop polls the connection map before the queued +// (setImmediate) removal runs, and nothing in ProxyClient retries after +// the single post-disconnect reconnection attempt (attemptConnection +// swallows refusals, so the retry helper returns after one cycle). The +// C++ port removes the connection deterministically, so this test +// asserts the portable intent: the connection drops when the proxy +// leaves, and re-issuing setProxies (the supported client API) restores +// it once the proxy is back. +TEST_F(ProxyConnectionsTest, ReconnectsIfProxyNodeGoesOfflineAndComesBack) { + blockingWait(this->proxiedNode->setProxies( + this->streamPartId, + {this->proxyNode1->getPeerDescriptor()}, + std::nullopt, + this->proxiedNodeUserId)); + EXPECT_TRUE(this->proxiedNode->hasStreamPart(this->streamPartId)); + EXPECT_TRUE(this->hasConnectionToProxy(this->proxyNode1->getNodeId())); + blockingWait(this->proxyNode1->leave(this->streamPartId)); + blockingWait(waitForCondition( + [this]() { + return !this->hasConnectionToProxy(this->proxyNode1->getNodeId()); + }, + defaultTimeout)); + EXPECT_FALSE(this->hasConnectionFromProxy(*this->proxyNode1)); + this->proxyNode1->getStack().getContentDeliveryManager().joinStreamPart( + this->streamPartId); + // The rejoin races the reconnection attempt (each refused attempt + // costs the 5 s RPC timeout), so keep re-issuing setProxies until + // the connection is back — the loop a real client's keep-alive + // logic runs. + const auto deadline = std::chrono::steady_clock::now() + reconnectTimeout; + bool reconnected = false; + while (std::chrono::steady_clock::now() < deadline) { + blockingWait(this->proxiedNode->setProxies( + this->streamPartId, + {this->proxyNode1->getPeerDescriptor()}, + std::nullopt, + this->proxiedNodeUserId)); + if (this->hasConnectionToProxy(this->proxyNode1->getNodeId())) { + reconnected = true; + break; + } + } + EXPECT_TRUE(reconnected); + blockingWait(waitForCondition( + [this]() { return this->hasConnectionFromProxy(*this->proxyNode1); }, + defaultTimeout)); +} + +TEST_F(ProxyConnectionsTest, BidirectionalProxiesCanPublish) { + blockingWait(this->proxiedNode->setProxies( + this->streamPartId, + {this->proxyNode1->getPeerDescriptor()}, + std::nullopt, + this->proxiedNodeUserId)); + EXPECT_TRUE(this->proxiedNode->hasStreamPart(this->streamPartId)); + this->waitForMessage(*this->proxyNode1, *this->proxiedNode); +} + +TEST_F(ProxyConnectionsTest, BidirectionalProxiesCanSubscribe) { + blockingWait(this->proxiedNode->setProxies( + this->streamPartId, + {this->proxyNode1->getPeerDescriptor()}, + std::nullopt, + this->proxiedNodeUserId)); + EXPECT_TRUE(this->proxiedNode->hasStreamPart(this->streamPartId)); + this->waitForMessage(*this->proxiedNode, *this->proxyNode1); +} + +TEST_F(ProxyConnectionsTest, CantJoinProxiedStreamPart) { + blockingWait(this->proxiedNode->setProxies( + this->streamPartId, + {this->proxyNode1->getPeerDescriptor()}, + ProxyDirection::PUBLISH, + this->proxiedNodeUserId)); + EXPECT_THROW( + blockingWait(this->proxiedNode->join(this->streamPartId)), + std::runtime_error); +} + +TEST_F(ProxyConnectionsTest, CantBroadcastToSubscribeOnlyProxiedStreamPart) { + blockingWait(this->proxiedNode->setProxies( + this->streamPartId, + {this->proxyNode1->getPeerDescriptor()}, + ProxyDirection::SUBSCRIBE, + this->proxiedNodeUserId)); + EXPECT_THROW( + blockingWait(this->proxiedNode->broadcast( + createTestMessage(this->streamPartId))), + std::runtime_error); +}