diff --git a/packages/streamr-dht/CMakeLists.txt b/packages/streamr-dht/CMakeLists.txt index f904becb..46370366 100644 --- a/packages/streamr-dht/CMakeLists.txt +++ b/packages/streamr-dht/CMakeLists.txt @@ -286,9 +286,34 @@ if(NOT IOS AND STREAMR_MODULES_SUPPORTED) PUBLIC GTest::gtest PUBLIC streamr-dht-test-main ) + + # End-to-end tests: real sockets on 127.0.0.1, nodes owning their + # ConnectionManagers (milestone B exit criterion). + add_executable(streamr-dht-test-end-to-end + test/end-to-end/Layer0Test.cpp + test/end-to-end/Layer0WebrtcTest.cpp + test/end-to-end/Layer0MixedConnectionTypesTest.cpp + test/end-to-end/Layer0Layer1Test.cpp + test/end-to-end/Layer0WebrtcLayer1Test.cpp + ) + target_include_directories(streamr-dht-test-end-to-end + PUBLIC $ + ) + streamr_enable_imports(streamr-dht-test-end-to-end) + target_link_libraries(streamr-dht-test-end-to-end + PUBLIC streamr-dht + PUBLIC streamr-dht-test-utils + PUBLIC streamr::streamr-logger + PUBLIC streamr::streamr-eventemitter + PUBLIC streamr::streamr-utils + PUBLIC streamr::streamr-proto-rpc + PUBLIC GTest::gtest + PUBLIC streamr-dht-test-main + ) if (NOT (${VCPKG_TARGET_TRIPLET} MATCHES "android")) include(GoogleTest) gtest_discover_tests(streamr-dht-test-unit) gtest_discover_tests(streamr-dht-test-integration) + gtest_discover_tests(streamr-dht-test-end-to-end) endif() endif() diff --git a/packages/streamr-dht/lint.sh b/packages/streamr-dht/lint.sh index 93bb4ee4..234328dd 100755 --- a/packages/streamr-dht/lint.sh +++ b/packages/streamr-dht/lint.sh @@ -75,6 +75,10 @@ echo "Running clangd-tidy on $FILES" # CodeLocation/MakeAndRegisterTestInfo reported ambiguous — like the other # excluded files; the compiler builds and runs it on every platform and # clang-format still checks it.) +# The end-to-end tests (phase B3) import the DhtNode aggregate like +# DhtNodeTest.cpp and trip the same std-type unification false positive +# (gtest CodeLocation/MakeAndRegisterTestInfo ambiguous); the compiler +# builds and runs all five, clang-format still checks them. # WebrtcConnectionTest.cpp, WebrtcConnectorTest.cpp and # RpcConnectionsOverWebrtcTest.cpp (phase B2) trip the same std-type # unification false positive (gtest CodeLocation/MakeAndRegisterTestInfo @@ -90,8 +94,12 @@ echo "Running clangd-tidy on $FILES" # ConnectionManager import sets; the compiler builds and runs all three, # clang-format still checks them. (ConnectivityCheckingTest.cpp imports # the same cluster but does NOT trip it.) -HEAVY_TIDY_FILES='./test/integration/MockLayer1Layer0Test.cpp ./test/integration/Layer1ScaleTest.cpp ./test/integration/DhtNodeExternalApiTest.cpp ./test/integration/KademliaCorrectnessTest.cpp' -TIDY_FILES=$(echo "$FILES" | tr ' ' '\n' | grep -v 'test/integration/ConnectionLockingTest.cpp' | grep -v 'test/unit/PendingConnectionTest.cpp' | grep -v 'test/unit/SimulatorTest.cpp' | grep -v 'test/integration/SimultaneousConnectionsTest.cpp' | grep -v 'test/integration/ConnectionManagerIntegrationTest.cpp' | grep -v 'test/unit/DhtNodeRpcLocalTest.cpp' | grep -v 'test/integration/DhtNodeRpcRemoteTest.cpp' | grep -v 'test/unit/RoutingSessionTest.cpp' | grep -v 'test/unit/RecursiveOperationSessionTest.cpp' | grep -v 'test/unit/StoreRpcLocalTest.cpp' | grep -v 'test/unit/StoreManagerTest.cpp' | grep -v 'test/integration/MultipleEntryPointJoiningTest.cpp' | grep -v 'test/utils/DhtNodeTestUtils.hpp' | grep -v 'test/integration/DhtNodeTest.cpp' | grep -v 'test/integration/MockLayer1Layer0Test.cpp' | grep -v 'test/integration/Layer1ScaleTest.cpp' | grep -v 'test/integration/DhtNodeExternalApiTest.cpp' | grep -v 'test/integration/KademliaCorrectnessTest.cpp' | grep -v 'test/unit/CreatePeerDescriptorTest.cpp' | grep -v 'test/unit/ConnectivityRequestHandlerTest.cpp' | grep -v 'test/integration/WebsocketConnectionManagementTest.cpp' | grep -v 'test/unit/WebrtcConnectionTest.cpp' | grep -v 'test/unit/WebrtcConnectorTest.cpp' | grep -v 'test/integration/RpcConnectionsOverWebrtcTest.cpp' | grep -v 'test/integration/WebrtcConnectorRpcTest.cpp' | grep -v 'test/integration/WebrtcConnectionManagementTest.cpp' | tr '\n' ' ') +# (MockLayer1Layer0Test.cpp moved from the own-process list to outright +# exclusion in phase B3: DhtNode's BMI grew with the ConnectionManager-path +# imports and the TU now trips the branded-ServiceID false positive +# documented for MultipleEntryPointJoiningTest above.) +HEAVY_TIDY_FILES='./test/integration/Layer1ScaleTest.cpp ./test/integration/DhtNodeExternalApiTest.cpp ./test/integration/KademliaCorrectnessTest.cpp' +TIDY_FILES=$(echo "$FILES" | tr ' ' '\n' | grep -v 'test/integration/ConnectionLockingTest.cpp' | grep -v 'test/unit/PendingConnectionTest.cpp' | grep -v 'test/unit/SimulatorTest.cpp' | grep -v 'test/integration/SimultaneousConnectionsTest.cpp' | grep -v 'test/integration/ConnectionManagerIntegrationTest.cpp' | grep -v 'test/unit/DhtNodeRpcLocalTest.cpp' | grep -v 'test/integration/DhtNodeRpcRemoteTest.cpp' | grep -v 'test/unit/RoutingSessionTest.cpp' | grep -v 'test/unit/RecursiveOperationSessionTest.cpp' | grep -v 'test/unit/StoreRpcLocalTest.cpp' | grep -v 'test/unit/StoreManagerTest.cpp' | grep -v 'test/integration/MultipleEntryPointJoiningTest.cpp' | grep -v 'test/utils/DhtNodeTestUtils.hpp' | grep -v 'test/integration/DhtNodeTest.cpp' | grep -v 'test/integration/MockLayer1Layer0Test.cpp' | grep -v 'test/integration/Layer1ScaleTest.cpp' | grep -v 'test/integration/DhtNodeExternalApiTest.cpp' | grep -v 'test/integration/KademliaCorrectnessTest.cpp' | grep -v 'test/unit/CreatePeerDescriptorTest.cpp' | grep -v 'test/unit/ConnectivityRequestHandlerTest.cpp' | grep -v 'test/integration/WebsocketConnectionManagementTest.cpp' | grep -v 'test/unit/WebrtcConnectionTest.cpp' | grep -v 'test/unit/WebrtcConnectorTest.cpp' | grep -v 'test/integration/RpcConnectionsOverWebrtcTest.cpp' | grep -v 'test/integration/WebrtcConnectorRpcTest.cpp' | grep -v 'test/integration/WebrtcConnectionManagementTest.cpp' | grep -v 'test/end-to-end/Layer0Test.cpp' | grep -v 'test/end-to-end/Layer0WebrtcTest.cpp' | grep -v 'test/end-to-end/Layer0MixedConnectionTypesTest.cpp' | grep -v 'test/end-to-end/Layer0Layer1Test.cpp' | grep -v 'test/end-to-end/Layer0WebrtcLayer1Test.cpp' | tr '\n' ' ') # Chunked: one clangd process accumulates source-location space across the # files it serves and never releases it; with the phase-A8 module graph a # single process no longer survives the whole list (mid-batch the affected diff --git a/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm b/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm index 08a41c03..812dc1da 100644 --- a/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm +++ b/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm @@ -128,8 +128,13 @@ private: std::recursive_mutex mMutex; public: + // Members initialize in DECLARATION order (options first), so `host` + // must read from the already-moved-into MEMBER, not the parameter — the + // old `host(options.host)` read the moved-from parameter and produced an + // empty host, which surfaced as a `ws://:port` connectivity URL once the + // end-to-end path used the advertised descriptor. explicit WebsocketServerConnector(WebsocketServerConnectorOptions&& options) - : host(options.host), options(std::move(options)) { + : options(std::move(options)), host(this->options.host) { if (this->options.portRange.has_value()) { this->websocketServer = std::make_unique(std::move( WebsocketServerConfig{ diff --git a/packages/streamr-dht/modules/dht/DhtNode.cppm b/packages/streamr-dht/modules/dht/DhtNode.cppm index 49c83637..1557240c 100644 --- a/packages/streamr-dht/modules/dht/DhtNode.cppm +++ b/packages/streamr-dht/modules/dht/DhtNode.cppm @@ -43,7 +43,12 @@ import streamr.utils.ExecutorHelper; import streamr.utils.waitForCondition; import streamr.logger.SLogger; import streamr.dht.ConnectionLocker; +import streamr.dht.ConnectionManager; +import streamr.dht.ConnectorFacade; import streamr.dht.ConnectionsView; +import streamr.dht.createPeerDescriptor; +import streamr.dht.PortRange; +import streamr.dht.webrtcTypes; import streamr.dht.consts; import streamr.dht.DhtCallContext; import streamr.dht.DhtNodeRpcLocal; @@ -135,13 +140,31 @@ struct DhtNodeOptions { std::optional neighborPingLimit; std::optional rpcRequestTimeout; - // Given transport (required in this phase). The three roles are the same - // object for a SimulatorTransport / ConnectionManager. + // Given transport. When null (TS: options.transport === undefined) the + // node creates and owns a ConnectionManager over a + // DefaultConnectorFacade using the connectivity options below, with + // ITSELF as the connectors' signalling transport (TS `transport: this`). Transport* transport = nullptr; ConnectionsView* connectionsView = nullptr; ConnectionLocker* connectionLocker = nullptr; // may be null std::optional peerDescriptor; std::vector entryPoints; + + // Connectivity options for the owned-ConnectionManager path (TS + // DhtNodeOptions). websocketServerEnableTls/TLS certificates and the + // GeoIP region fallback stay deferred (milestones D/E); region defaults + // to 0 when not given. + std::optional websocketHost; + std::optional websocketPortRange; + std::optional nodeId; + std::optional region; + std::vector iceServers; + std::optional webrtcAllowPrivateAddresses; + std::optional webrtcDatachannelBufferThresholdLow; + std::optional webrtcDatachannelBufferThresholdHigh; + std::optional webrtcPortRange; + std::optional externalIp; + std::optional maxMessageSize; }; class DhtNode : public Transport { @@ -154,6 +177,10 @@ private: LocalDataStore localDataStore; std::unique_ptr rpcCommunicator; Transport* transportPtr = nullptr; + // Set only on the owned-ConnectionManager path (options.transport null); + // stop() stops it last, like TS. + std::shared_ptr + ownedConnectionManager; ConnectionsView* connectionsView = nullptr; ConnectionLocker* connectionLocker = nullptr; std::shared_ptr @@ -411,14 +438,64 @@ public: co_return; } this->started = true; - if (this->options.transport == nullptr) { - throw std::runtime_error( - "DhtNode requires a transport in this build (the " - "ConnectionManager path arrives in milestone B)"); + if (this->options.transport != nullptr) { + this->transportPtr = this->options.transport; + this->connectionsView = this->options.connectionsView; + this->connectionLocker = this->options.connectionLocker; + } else { + // TS: no transport given — create and own a ConnectionManager + // whose connectors signal through THIS node (routed messages). + using streamr::dht::connection::ConnectionManager; + using streamr::dht::connection::ConnectionManagerOptions; + using streamr::dht::connection::DefaultConnectorFacade; + using streamr::dht::connection::DefaultConnectorFacadeOptions; + DefaultConnectorFacadeOptions facadeOptions{ + .transport = *this, + .entryPoints = this->options.entryPoints, + .iceServers = this->options.iceServers, + .webrtcAllowPrivateAddresses = + this->options.webrtcAllowPrivateAddresses, + .webrtcDatachannelBufferThresholdLow = + this->options.webrtcDatachannelBufferThresholdLow, + .webrtcDatachannelBufferThresholdHigh = + this->options.webrtcDatachannelBufferThresholdHigh, + .externalIp = this->options.externalIp, + .webrtcPortRange = this->options.webrtcPortRange, + .maxMessageSize = this->options.maxMessageSize, + .createLocalPeerDescriptor = + [this](const ::dht::ConnectivityResponse& response) { + return this->generatePeerDescriptorCallBack(response); + }}; + // If an own PeerDescriptor with a websocket is given, serve on + // exactly that port; else use the configured port range (the + // websocketHost may be undefined then). (TS DhtNode lines + // 231-243.) + if (this->options.peerDescriptor.has_value() && + this->options.peerDescriptor->has_websocket()) { + facadeOptions.websocketHost = + this->options.peerDescriptor->websocket().host(); + const auto port = static_cast( + this->options.peerDescriptor->websocket().port()); + facadeOptions.websocketPortRange = + streamr::dht::types::PortRange{.min = port, .max = port}; + } else if (this->options.websocketPortRange.has_value()) { + facadeOptions.websocketHost = this->options.websocketHost; + facadeOptions.websocketPortRange = + this->options.websocketPortRange; + } + this->ownedConnectionManager = + std::make_shared(ConnectionManagerOptions{ + .createConnectorFacade = [facadeOptions = + std::move(facadeOptions)]() + -> std::shared_ptr { + return std::make_shared( + facadeOptions); + }}); + this->ownedConnectionManager->start(); + this->transportPtr = this->ownedConnectionManager.get(); + this->connectionsView = this->ownedConnectionManager.get(); + this->connectionLocker = this->ownedConnectionManager.get(); } - this->transportPtr = this->options.transport; - this->connectionsView = this->options.connectionsView; - this->connectionLocker = this->options.connectionLocker; if (this->connectionLocker != nullptr) { this->connectionLockerShared = std::shared_ptr( this->connectionLocker, [](ConnectionLocker*) {}); @@ -550,6 +627,23 @@ public: return *this->localPeerDescriptor; } + // TS generatePeerDescriptorCallBack: build (or adopt) the local peer + // descriptor from the connectivity response. The GeoIP/CDN region + // fallbacks are deferred (milestone E) — region comes from options or + // defaults to 0. + PeerDescriptor generatePeerDescriptorCallBack( + const ::dht::ConnectivityResponse& connectivityResponse) { + if (this->options.peerDescriptor.has_value()) { + this->localPeerDescriptor = this->options.peerDescriptor; + } else { + const uint32_t region = this->options.region.value_or(0); + this->localPeerDescriptor = + streamr::dht::helpers::createPeerDescriptor( + connectivityResponse, region, this->options.nodeId); + } + return this->localPeerDescriptor.value(); + } + void stop() override { if (this->abortController.getSignal().aborted || !this->started) { return; @@ -571,7 +665,18 @@ public: } this->router->stop(); this->recursiveOperationManager->stop(); - // The transport was given, not created here, so it is not stopped. + if (this->ownedConnectionManager) { + // The transport was created in start(), so this component is + // responsible for stopping it (TS: options.transport + // undefined). A transport given in options is NOT stopped. + // The shared_ptr is deliberately KEPT (and transportPtr left + // pointing at it): straggler detached tasks are only drained at + // the communicators' destruction, and a send into a STOPPED + // ConnectionManager is a guarded no-op while a send into a + // destroyed one is a use-after-free (TS keeps the stopped + // object alive through GC). + this->ownedConnectionManager->stop(); + } this->connectionLocker = nullptr; } diff --git a/packages/streamr-dht/test/end-to-end/Layer0Layer1Test.cpp b/packages/streamr-dht/test/end-to-end/Layer0Layer1Test.cpp new file mode 100644 index 00000000..5963203b --- /dev/null +++ b/packages/streamr-dht/test/end-to-end/Layer0Layer1Test.cpp @@ -0,0 +1,144 @@ +// Ported from packages/dht/test/end-to-end/Layer0-Layer1.test.ts +// (v103.8.0-rc.3): two stream (layer-1) DHTs stacked on layer-0 nodes that +// talk over real websockets; each stream node's single closest contact must +// be the OTHER member of its stream. +#include +#include +#include +#include "packages/dht/protos/DhtRpc.pb.h" + +#include // IWYU pragma: keep + +import streamr.dht.DhtNode; +import streamr.dht.Identifiers; +import streamr.dht.PortRange; +import streamr.dht.TestUtils; +import streamr.dht.protos; +import streamr.utils.CoroutineHelper; + +using ::dht::PeerDescriptor; +using streamr::dht::DhtNode; +using streamr::dht::DhtNodeOptions; +using streamr::dht::Identifiers; +using streamr::dht::numberOfNodesPerKBucketDefault; +using streamr::dht::ServiceID; +using streamr::dht::testutils::createMockPeerDescriptor; +using streamr::dht::types::PortRange; +using streamr::utils::blockingWait; + +namespace { +constexpr uint16_t entryPointPort = 10016; +constexpr PortRange websocketPortRange{.min = 10017, .max = 10018}; +} // namespace + +class Layer0Layer1Test : public ::testing::Test { +protected: + PeerDescriptor epPeerDescriptor; + std::shared_ptr epDhtNode; + std::shared_ptr node1; + std::shared_ptr node2; + std::shared_ptr stream1Node1; + std::shared_ptr stream1Node2; + std::shared_ptr stream2Node1; + std::shared_ptr stream2Node2; + + void SetUp() override { + this->epPeerDescriptor = createMockPeerDescriptor(); + this->epPeerDescriptor.mutable_websocket()->set_host("127.0.0.1"); + this->epPeerDescriptor.mutable_websocket()->set_port(entryPointPort); + this->epPeerDescriptor.mutable_websocket()->set_tls(false); + + this->epDhtNode = std::make_shared( + DhtNodeOptions{.peerDescriptor = this->epPeerDescriptor}); + blockingWait(this->epDhtNode->start()); + blockingWait(this->epDhtNode->joinDht({this->epPeerDescriptor})); + + this->node1 = std::make_shared(DhtNodeOptions{ + .entryPoints = {this->epPeerDescriptor}, + .websocketPortRange = websocketPortRange}); + this->node2 = std::make_shared(DhtNodeOptions{ + .entryPoints = {this->epPeerDescriptor}, + .websocketPortRange = websocketPortRange}); + blockingWait(this->node1->start()); + blockingWait(this->node2->start()); + + this->stream1Node1 = std::make_shared(DhtNodeOptions{ + .serviceId = ServiceID{"stream1"}, + .transport = this->epDhtNode.get(), + .connectionsView = this->epDhtNode->getConnectionsView()}); + this->stream1Node2 = std::make_shared(DhtNodeOptions{ + .serviceId = ServiceID{"stream1"}, + .transport = this->node1.get(), + .connectionsView = this->node1->getConnectionsView()}); + this->stream2Node1 = std::make_shared(DhtNodeOptions{ + .serviceId = ServiceID{"stream2"}, + .transport = this->epDhtNode.get(), + .connectionsView = this->epDhtNode->getConnectionsView()}); + this->stream2Node2 = std::make_shared(DhtNodeOptions{ + .serviceId = ServiceID{"stream2"}, + .transport = this->node2.get(), + .connectionsView = this->node2->getConnectionsView()}); + std::vector> starts; + starts.push_back(this->stream1Node1->start()); + starts.push_back(this->stream1Node2->start()); + starts.push_back(this->stream2Node1->start()); + starts.push_back(this->stream2Node2->start()); + blockingWait(folly::coro::collectAllRange(std::move(starts))); + } + + void TearDown() override { + this->node1->stop(); + this->node2->stop(); + this->epDhtNode->stop(); + this->stream1Node1->stop(); + this->stream1Node2->stop(); + this->stream2Node1->stop(); + this->stream2Node2->stop(); + } +}; + +TEST_F(Layer0Layer1Test, HappyPath) { + { + std::vector> joins; + joins.push_back(this->node1->joinDht({this->epPeerDescriptor})); + joins.push_back(this->node2->joinDht({this->epPeerDescriptor})); + blockingWait(folly::coro::collectAllRange(std::move(joins))); + } + { + std::vector> joins; + joins.push_back(this->stream1Node1->joinDht({this->epPeerDescriptor})); + joins.push_back(this->stream1Node2->joinDht({this->epPeerDescriptor})); + blockingWait(folly::coro::collectAllRange(std::move(joins))); + } + { + std::vector> joins; + joins.push_back(this->stream2Node1->joinDht({this->epPeerDescriptor})); + joins.push_back(this->stream2Node2->joinDht({this->epPeerDescriptor})); + blockingWait(folly::coro::collectAllRange(std::move(joins))); + } + + const auto contacts1 = + this->stream1Node1->getClosestContacts(numberOfNodesPerKBucketDefault); + const auto contacts2 = + this->stream1Node2->getClosestContacts(numberOfNodesPerKBucketDefault); + const auto contacts3 = + this->stream2Node1->getClosestContacts(numberOfNodesPerKBucketDefault); + const auto contacts4 = + this->stream2Node2->getClosestContacts(numberOfNodesPerKBucketDefault); + ASSERT_EQ(contacts1.size(), 1); + ASSERT_EQ(contacts2.size(), 1); + ASSERT_EQ(contacts3.size(), 1); + ASSERT_EQ(contacts4.size(), 1); + EXPECT_TRUE( + Identifiers::areEqualPeerDescriptors( + contacts1[0], this->node1->getLocalPeerDescriptor())); + EXPECT_TRUE( + Identifiers::areEqualPeerDescriptors( + contacts2[0], this->epDhtNode->getLocalPeerDescriptor())); + EXPECT_TRUE( + Identifiers::areEqualPeerDescriptors( + contacts3[0], this->node2->getLocalPeerDescriptor())); + EXPECT_TRUE( + Identifiers::areEqualPeerDescriptors( + contacts4[0], this->epDhtNode->getLocalPeerDescriptor())); +} diff --git a/packages/streamr-dht/test/end-to-end/Layer0MixedConnectionTypesTest.cpp b/packages/streamr-dht/test/end-to-end/Layer0MixedConnectionTypesTest.cpp new file mode 100644 index 00000000..a6548220 --- /dev/null +++ b/packages/streamr-dht/test/end-to-end/Layer0MixedConnectionTypesTest.cpp @@ -0,0 +1,126 @@ +// Ported from packages/dht/test/end-to-end/Layer0MixedConnectionTypes.test.ts +// (v103.8.0-rc.3): a mixed network — the entry point and two nodes run real +// websocket servers, three nodes are serverless (websocket-client towards +// the servers, WebRTC among themselves). Adaptation: the TS +// waitForEvent(transport, 'connected') gates in the first case become +// waitForCondition on getConnectionCount() >= 1 for the serverless joiners +// (the same "has connected to somebody" postcondition). +#include +#include +#include +#include +#include "packages/dht/protos/DhtRpc.pb.h" + +#include // IWYU pragma: keep + +import streamr.dht.DhtNode; +import streamr.dht.PortRange; +import streamr.dht.TestUtils; +import streamr.dht.protos; +import streamr.utils.CoroutineHelper; +import streamr.utils.waitForCondition; + +using ::dht::PeerDescriptor; +using streamr::dht::DhtNode; +using streamr::dht::DhtNodeOptions; +using streamr::dht::testutils::createMockPeerDescriptor; +using streamr::dht::types::PortRange; +using streamr::utils::blockingWait; +using streamr::utils::waitForCondition; + +namespace { +constexpr uint16_t entryPointPort = 11221; +constexpr PortRange websocketPortRange{.min = 11222, .max = 11223}; +constexpr size_t serverNodeCount = 2; +constexpr size_t serverlessNodeCount = 3; +constexpr auto connectTimeout = std::chrono::seconds(30); +constexpr auto pollInterval = std::chrono::milliseconds(200); +constexpr size_t kBucketSize = 2; +} // namespace + +class Layer0MixedConnectionTypesTest : public ::testing::Test { +protected: + PeerDescriptor epPeerDescriptor; + std::shared_ptr epDhtNode; + // nodes[0..1] run websocket servers, nodes[2..4] are serverless. + std::vector> nodes; + + void SetUp() override { + this->epPeerDescriptor = createMockPeerDescriptor(); + this->epPeerDescriptor.mutable_websocket()->set_host("127.0.0.1"); + this->epPeerDescriptor.mutable_websocket()->set_port(entryPointPort); + this->epPeerDescriptor.mutable_websocket()->set_tls(false); + + this->epDhtNode = std::make_shared(DhtNodeOptions{ + .numberOfNodesPerKBucket = kBucketSize, + .peerDescriptor = this->epPeerDescriptor}); + blockingWait(this->epDhtNode->start()); + blockingWait(this->epDhtNode->joinDht({this->epPeerDescriptor})); + + for (size_t i = 0; i < serverNodeCount; i++) { + this->nodes.push_back( + std::make_shared(DhtNodeOptions{ + .entryPoints = {this->epPeerDescriptor}, + .websocketPortRange = websocketPortRange})); + } + for (size_t i = 0; i < serverlessNodeCount; i++) { + this->nodes.push_back( + std::make_shared( + DhtNodeOptions{.entryPoints = {this->epPeerDescriptor}})); + } + std::vector> starts; + starts.reserve(this->nodes.size()); + for (const auto& node : this->nodes) { + starts.push_back(node->start()); + } + blockingWait(folly::coro::collectAllRange(std::move(starts))); + blockingWait(this->epDhtNode->joinDht({this->epPeerDescriptor})); + } + + void TearDown() override { + this->epDhtNode->stop(); + for (const auto& node : this->nodes) { + node->stop(); + } + } +}; + +TEST_F(Layer0MixedConnectionTypesTest, TwoNonServerPeersJoinFirst) { + { + std::vector> joins; + joins.push_back(this->nodes[2]->joinDht({this->epPeerDescriptor})); + joins.push_back(this->nodes[3]->joinDht({this->epPeerDescriptor})); + blockingWait(folly::coro::collectAllRange(std::move(joins))); + for (size_t i : {size_t{2}, size_t{3}}) { + auto* view = this->nodes[i]->getConnectionsView(); + EXPECT_NO_THROW(blockingWait(waitForCondition( + [view]() { return view->getConnectionCount() >= 1; }, + connectTimeout, + pollInterval))); + } + } + { + std::vector> joins; + joins.push_back(this->nodes[0]->joinDht({this->epPeerDescriptor})); + joins.push_back(this->nodes[1]->joinDht({this->epPeerDescriptor})); + joins.push_back(this->nodes[4]->joinDht({this->epPeerDescriptor})); + blockingWait(folly::coro::collectAllRange(std::move(joins))); + } + EXPECT_GE(this->nodes[0]->getNeighborCount(), 2); + EXPECT_GE(this->nodes[1]->getNeighborCount(), 2); + EXPECT_GE(this->nodes[2]->getNeighborCount(), 2); + EXPECT_GE(this->nodes[3]->getNeighborCount(), 2); + EXPECT_GE(this->nodes[4]->getNeighborCount(), 1); +} + +TEST_F(Layer0MixedConnectionTypesTest, SimultaneousJoins) { + std::vector> joins; + joins.reserve(this->nodes.size()); + for (const auto& node : this->nodes) { + joins.push_back(node->joinDht({this->epPeerDescriptor})); + } + blockingWait(folly::coro::collectAllRange(std::move(joins))); + for (const auto& node : this->nodes) { + EXPECT_GE(node->getNeighborCount(), 2); + } +} diff --git a/packages/streamr-dht/test/end-to-end/Layer0Test.cpp b/packages/streamr-dht/test/end-to-end/Layer0Test.cpp new file mode 100644 index 00000000..5b68960e --- /dev/null +++ b/packages/streamr-dht/test/end-to-end/Layer0Test.cpp @@ -0,0 +1,82 @@ +// Ported from packages/dht/test/end-to-end/Layer0.test.ts (v103.8.0-rc.3): +// five DhtNodes over REAL websocket servers on 127.0.0.1 (no transport given +// — each node creates and owns its ConnectionManager, checks its own +// connectivity and generates a signed peer descriptor), the four late nodes +// joining through the entry point. Adaptation: the jest Promise.all becomes +// folly collectAllRange over the start()/joinDht() tasks. +#include +#include +#include +#include "packages/dht/protos/DhtRpc.pb.h" + +#include // IWYU pragma: keep + +import streamr.dht.DhtNode; +import streamr.dht.PortRange; +import streamr.dht.protos; +import streamr.utils.CoroutineHelper; + +using ::dht::PeerDescriptor; +using streamr::dht::DhtNode; +using streamr::dht::DhtNodeOptions; +using streamr::dht::types::PortRange; +using streamr::utils::blockingWait; + +namespace { + +constexpr uint16_t entryPointPort = 10011; +constexpr PortRange websocketPortRange{.min = 10012, .max = 10015}; +constexpr size_t nodeCount = 4; + +} // namespace + +class Layer0Test : public ::testing::Test { +protected: + std::shared_ptr epDhtNode; + std::vector> nodes; + + void SetUp() override { + this->epDhtNode = std::make_shared(DhtNodeOptions{ + .websocketHost = "127.0.0.1", + .websocketPortRange = + PortRange{.min = entryPointPort, .max = entryPointPort}}); + blockingWait(this->epDhtNode->start()); + blockingWait(this->epDhtNode->joinDht( + {this->epDhtNode->getLocalPeerDescriptor()})); + + for (size_t i = 0; i < nodeCount; i++) { + this->nodes.push_back( + std::make_shared(DhtNodeOptions{ + .entryPoints = {this->epDhtNode->getLocalPeerDescriptor()}, + .websocketHost = "127.0.0.1", + .websocketPortRange = websocketPortRange})); + } + std::vector> starts; + starts.reserve(nodeCount); + for (const auto& node : this->nodes) { + starts.push_back(node->start()); + } + blockingWait(folly::coro::collectAllRange(std::move(starts))); + } + + void TearDown() override { + for (const auto& node : this->nodes) { + node->stop(); + } + this->epDhtNode->stop(); + } +}; + +TEST_F(Layer0Test, HappyPath) { + std::vector> joins; + joins.reserve(nodeCount); + for (const auto& node : this->nodes) { + joins.push_back( + node->joinDht({this->epDhtNode->getLocalPeerDescriptor()})); + } + blockingWait(folly::coro::collectAllRange(std::move(joins))); + + for (const auto& node : this->nodes) { + EXPECT_GE(node->getNeighborCount(), 2); + } +} diff --git a/packages/streamr-dht/test/end-to-end/Layer0WebrtcLayer1Test.cpp b/packages/streamr-dht/test/end-to-end/Layer0WebrtcLayer1Test.cpp new file mode 100644 index 00000000..ff2f2291 --- /dev/null +++ b/packages/streamr-dht/test/end-to-end/Layer0WebrtcLayer1Test.cpp @@ -0,0 +1,113 @@ +// Ported from packages/dht/test/end-to-end/Layer0Webrtc-Layer1.test.ts +// (v103.8.0-rc.3): a layer-1 DHT stacked on a layer-0 whose non-entry-point +// members are all serverless, so every layer-0 connection between them is a +// real WebRTC data channel; each layer-1 node shares its layer-0 node's id +// (the TS nodeId option). +#include +#include +#include +#include "packages/dht/protos/DhtRpc.pb.h" + +#include // IWYU pragma: keep + +import streamr.dht.DhtNode; +import streamr.dht.Identifiers; +import streamr.dht.TestUtils; +import streamr.dht.protos; +import streamr.utils.CoroutineHelper; + +using ::dht::PeerDescriptor; +using streamr::dht::DhtNode; +using streamr::dht::DhtNodeOptions; +using streamr::dht::Identifiers; +using streamr::dht::ServiceID; +using streamr::dht::testutils::createMockPeerDescriptor; +using streamr::utils::blockingWait; + +namespace { +constexpr uint16_t entryPointPort = 23232; +constexpr size_t nodeCount = 4; +} // namespace + +class Layer0WebrtcLayer1Test : public ::testing::Test { +protected: + PeerDescriptor entrypointDescriptor; + std::shared_ptr layer0EntryPoint; + std::shared_ptr layer1EntryPoint; + std::vector> layer0Nodes; + std::vector> layer1Nodes; + + void SetUp() override { + this->entrypointDescriptor = createMockPeerDescriptor(); + this->entrypointDescriptor.mutable_websocket()->set_host("127.0.0.1"); + this->entrypointDescriptor.mutable_websocket()->set_port( + entryPointPort); + this->entrypointDescriptor.mutable_websocket()->set_tls(false); + + this->layer0EntryPoint = std::make_shared( + DhtNodeOptions{.peerDescriptor = this->entrypointDescriptor}); + blockingWait(this->layer0EntryPoint->start()); + + std::vector nodeIds; + for (size_t i = 0; i < nodeCount; i++) { + const auto nodeId = Identifiers::createRandomDhtAddress(); + nodeIds.push_back(nodeId); + this->layer0Nodes.push_back( + std::make_shared(DhtNodeOptions{ + .entryPoints = {this->entrypointDescriptor}, + .nodeId = nodeId})); + blockingWait(this->layer0Nodes.back()->start()); + } + + this->layer1EntryPoint = std::make_shared(DhtNodeOptions{ + .serviceId = ServiceID{"layer1"}, + .transport = this->layer0EntryPoint.get(), + .connectionsView = this->layer0EntryPoint->getConnectionsView(), + .nodeId = Identifiers::getNodeIdFromPeerDescriptor( + this->entrypointDescriptor)}); + blockingWait(this->layer1EntryPoint->start()); + for (size_t i = 0; i < nodeCount; i++) { + this->layer1Nodes.push_back( + std::make_shared(DhtNodeOptions{ + .serviceId = ServiceID{"layer1"}, + .transport = this->layer0Nodes[i].get(), + .connectionsView = + this->layer0Nodes[i]->getConnectionsView(), + .nodeId = nodeIds[i]})); + blockingWait(this->layer1Nodes.back()->start()); + } + + blockingWait( + this->layer0EntryPoint->joinDht({this->entrypointDescriptor})); + blockingWait( + this->layer1EntryPoint->joinDht({this->entrypointDescriptor})); + } + + void TearDown() override { + this->layer0EntryPoint->stop(); + for (const auto& node : this->layer0Nodes) { + node->stop(); + } + this->layer1EntryPoint->stop(); + for (const auto& node : this->layer1Nodes) { + node->stop(); + } + } +}; + +TEST_F(Layer0WebrtcLayer1Test, HappyPath) { + { + std::vector> joins; + joins.reserve(nodeCount); + for (const auto& node : this->layer0Nodes) { + joins.push_back(node->joinDht({this->entrypointDescriptor})); + } + blockingWait(folly::coro::collectAllRange(std::move(joins))); + } + for (const auto& node : this->layer1Nodes) { + blockingWait(node->joinDht({this->entrypointDescriptor})); + } + for (const auto& node : this->layer1Nodes) { + EXPECT_GE(node->getNeighborCount(), 2); + } +} diff --git a/packages/streamr-dht/test/end-to-end/Layer0WebrtcTest.cpp b/packages/streamr-dht/test/end-to-end/Layer0WebrtcTest.cpp new file mode 100644 index 00000000..e77fee3a --- /dev/null +++ b/packages/streamr-dht/test/end-to-end/Layer0WebrtcTest.cpp @@ -0,0 +1,108 @@ +// Ported from packages/dht/test/end-to-end/Layer0Webrtc.test.ts +// (v103.8.0-rc.3): the entry point serves a real websocket (given up-front +// in its peer descriptor); the four other nodes have NO websocket at all — +// they run their connectivity checks against the entry point over real +// sockets, connect to it as websocket clients, and to EACH OTHER over real +// WebRTC data channels signalled through the DHT. Adaptation: the TS +// waitForEvent('connected', matching node1) becomes waitForCondition on +// hasConnection in both directions (same postcondition the test asserts). +#include +#include +#include +#include +#include "packages/dht/protos/DhtRpc.pb.h" + +#include // IWYU pragma: keep + +import streamr.dht.DhtNode; +import streamr.dht.Identifiers; +import streamr.dht.TestUtils; +import streamr.dht.protos; +import streamr.utils.CoroutineHelper; +import streamr.utils.waitForCondition; + +using ::dht::PeerDescriptor; +using streamr::dht::DhtNode; +using streamr::dht::DhtNodeOptions; +using streamr::dht::Identifiers; +using streamr::dht::testutils::createMockPeerDescriptor; +using streamr::utils::blockingWait; +using streamr::utils::waitForCondition; + +namespace { +constexpr uint16_t entryPointPort = 10029; +constexpr size_t nodeCount = 4; +// TS gives the connected-wait 20 s within a 60 s jest budget. +constexpr auto connectTimeout = std::chrono::seconds(30); +constexpr auto pollInterval = std::chrono::milliseconds(200); +} // namespace + +class Layer0WebrtcTest : public ::testing::Test { +protected: + PeerDescriptor epPeerDescriptor; + std::shared_ptr epDhtNode; + std::vector> nodes; + + void SetUp() override { + this->epPeerDescriptor = createMockPeerDescriptor(); + this->epPeerDescriptor.mutable_websocket()->set_host("127.0.0.1"); + this->epPeerDescriptor.mutable_websocket()->set_port(entryPointPort); + this->epPeerDescriptor.mutable_websocket()->set_tls(false); + + this->epDhtNode = std::make_shared( + DhtNodeOptions{.peerDescriptor = this->epPeerDescriptor}); + blockingWait(this->epDhtNode->start()); + blockingWait(this->epDhtNode->joinDht({this->epPeerDescriptor})); + + for (size_t i = 0; i < nodeCount; i++) { + this->nodes.push_back( + std::make_shared( + DhtNodeOptions{.entryPoints = {this->epPeerDescriptor}})); + } + std::vector> starts; + starts.reserve(nodeCount); + for (const auto& node : this->nodes) { + starts.push_back(node->start()); + } + blockingWait(folly::coro::collectAllRange(std::move(starts))); + } + + void TearDown() override { + for (const auto& node : this->nodes) { + node->stop(); + } + this->epDhtNode->stop(); + } + + void expectMutuallyConnected(DhtNode& node1, DhtNode& node2) { + const auto nodeId1 = Identifiers::getNodeIdFromPeerDescriptor( + node1.getLocalPeerDescriptor()); + const auto nodeId2 = Identifiers::getNodeIdFromPeerDescriptor( + node2.getLocalPeerDescriptor()); + EXPECT_NO_THROW(blockingWait(waitForCondition( + [&node1, &node2, nodeId1, nodeId2]() { + return node1.getConnectionsView()->hasConnection(nodeId2) && + node2.getConnectionsView()->hasConnection(nodeId1); + }, + connectTimeout, + pollInterval))); + } +}; + +TEST_F(Layer0WebrtcTest, HappyPathTwoPeers) { + std::vector> joins; + joins.push_back(this->nodes[0]->joinDht({this->epPeerDescriptor})); + joins.push_back(this->nodes[1]->joinDht({this->epPeerDescriptor})); + blockingWait(folly::coro::collectAllRange(std::move(joins))); + this->expectMutuallyConnected(*this->nodes[0], *this->nodes[1]); +} + +TEST_F(Layer0WebrtcTest, HappyPathSimultaneousJoins) { + std::vector> joins; + joins.reserve(nodeCount); + for (const auto& node : this->nodes) { + joins.push_back(node->joinDht({this->epPeerDescriptor})); + } + blockingWait(folly::coro::collectAllRange(std::move(joins))); + this->expectMutuallyConnected(*this->nodes[0], *this->nodes[1]); +}