Phase B3 (part 1): DhtNode over its own ConnectionManager + Layer0 end-to-end tests

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 $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/test/utils>
+    )
+    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()

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

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<WebsocketServer>(std::move(
                 WebsocketServerConfig{

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<size_t> neighborPingLimit;
     std::optional<std::chrono::milliseconds> 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> peerDescriptor;
     std::vector<PeerDescriptor> 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<std::string> websocketHost;
+    std::optional<streamr::dht::types::PortRange> websocketPortRange;
+    std::optional<DhtAddress> nodeId;
+    std::optional<uint32_t> region;
+    std::vector<streamr::dht::connection::webrtc::IceServer> iceServers;
+    std::optional<bool> webrtcAllowPrivateAddresses;
+    std::optional<size_t> webrtcDatachannelBufferThresholdLow;
+    std::optional<size_t> webrtcDatachannelBufferThresholdHigh;
+    std::optional<streamr::dht::types::PortRange> webrtcPortRange;
+    std::optional<std::string> externalIp;
+    std::optional<size_t> maxMessageSize;
 };
 
 class DhtNode : public Transport {
@@ -154,6 +177,10 @@ private:
     LocalDataStore localDataStore;
     std::unique_ptr<RoutingRpcCommunicator> rpcCommunicator;
     Transport* transportPtr = nullptr;
+    // Set only on the owned-ConnectionManager path (options.transport null);
+    // stop() stops it last, like TS.
+    std::shared_ptr<streamr::dht::connection::ConnectionManager>
+        ownedConnectionManager;
     ConnectionsView* connectionsView = nullptr;
     ConnectionLocker* connectionLocker = nullptr;
     std::shared_ptr<ConnectionLocker>
@@ -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<uint16_t>(
+                    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<ConnectionManager>(ConnectionManagerOptions{
+                    .createConnectorFacade = [facadeOptions =
+                                                  std::move(facadeOptions)]()
+                        -> std::shared_ptr<DefaultConnectorFacade> {
+                        return std::make_shared<DefaultConnectorFacade>(
+                            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<ConnectionLocker>(
                 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;
     }