Phase B1: connectivity checking and signed peer descriptors

packages/streamr-dht/CMakeLists.txt

@@ -171,6 +171,8 @@ if(NOT IOS AND STREAMR_MODULES_SUPPORTED)
         test/unit/ConnectionManagerTest.cpp
         test/unit/WebsocketClientConnectorRpcRemoteTest.cpp
         test/unit/IdentifiersTest.cpp
+        test/unit/CreatePeerDescriptorTest.cpp
+        test/unit/ConnectivityRequestHandlerTest.cpp
         test/unit/ConnectionLockRpcRemoteTest.cpp
         test/unit/ConnectorFacadeTest.cpp
         test/unit/ConnectionsViewTest.cpp
@@ -260,6 +262,8 @@ if(NOT IOS AND STREAMR_MODULES_SUPPORTED)
         test/integration/Layer1ScaleTest.cpp
         test/integration/DhtNodeExternalApiTest.cpp
         test/integration/KademliaCorrectnessTest.cpp
+        test/integration/ConnectivityCheckingTest.cpp
+        test/integration/WebsocketConnectionManagementTest.cpp
     )
 
     target_include_directories(streamr-dht-test-integration

packages/streamr-dht/lint.sh

@@ -75,8 +75,16 @@ 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.)
+# CreatePeerDescriptorTest.cpp, ConnectivityRequestHandlerTest.cpp and
+# WebsocketConnectionManagementTest.cpp (phase B1) trip the same std-type
+# unification false positive (std::string method calls on protobuf getters
+# reported with mismatched string identities; gtest CodeLocation/
+# MakeAndRegisterTestInfo ambiguous) through the connectivityChecker /
+# 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' | tr '\n' ' ')
+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' | 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/ConnectorFacade.cppm

@@ -7,6 +7,8 @@ module;
 
 #include <chrono>
 #include <memory>
+#include <optional>
+#include <vector>
 
 #include <string>
 
@@ -68,7 +70,7 @@ struct DefaultConnectorFacadeOptions {
     Transport& transport;
     std::optional<std::string> websocketHost = std::nullopt;
     std::optional<PortRange> websocketPortRange = std::nullopt;
-    // std::vector<PeerDescriptor> entryPoints;
+    std::optional<std::vector<PeerDescriptor>> entryPoints = std::nullopt;
     // std::vector<IceServer> iceServers;
     // bool webrtcAllowPrivateAddresses;
     // int webrtcDatachannelBufferThresholdLow;
@@ -136,7 +138,7 @@ public:
             .portRange = this->options.websocketPortRange,
             .maxMessageSize = this->options.maxMessageSize,
             .host = this->options.websocketHost,
-            //.entrypoints = options.entryPoints,
+            .entrypoints = this->options.entryPoints,
             //.tlsCertificate = options.tlsCertificate,
             .serverEnableTls = false,
             //.autoCertifierUrl = options.autoCertifierUrl,
@@ -172,7 +174,11 @@ public:
                 peerDescriptor)) {
             return this->websocketClientConnector->connect(peerDescriptor);
         }
-
+        if (this->websocketServerConnector->isPossibleToFormConnection(
+                peerDescriptor)) {
+            return this->websocketServerConnector->connect(peerDescriptor);
+        }
+        // TS falls through to the WebrtcConnector here (milestone B2).
         return nullptr;
     }
 
@@ -184,7 +190,11 @@ public:
             return this->websocketClientConnector->connect(
                 peerDescriptor, std::move(errorCallback));
         }
-
+        if (this->websocketServerConnector->isPossibleToFormConnection(
+                peerDescriptor)) {
+            return this->websocketServerConnector->connect(peerDescriptor);
+        }
+        // TS falls through to the WebrtcConnector here (milestone B2).
         return nullptr;
     }
 

packages/streamr-dht/modules/connection/connectivityChecker.cppm

@@ -0,0 +1,160 @@
+// Module streamr.dht.connectivityChecker
+// Ported from packages/dht/src/connection/connectivityChecker.ts
+// (v103.8.0-rc.3): the client side of the NAT connectivity check. connectSync
+// opens a raw websocket client connection and waits for it to come up (the TS
+// connectAsync); sendConnectivityRequest connects to an entry point's
+// websocket server with the connectivityRequest action, sends a
+// ConnectivityRequest and waits for the ConnectivityResponse.
+//
+// Deliberately synchronous-blocking (a C++ design point): both entry points
+// run on dedicated worker threads — the connectivity-request handler's
+// executor on the server side, and the connector facade's start path on the
+// client side — never on an event-dispatch or RPC-delivery thread. The
+// event listeners are registered BEFORE connect()/send() fire, so a fast
+// completion cannot be lost (the TS code gets the same ordering from its
+// eager promises).
+module;
+
+#include <atomic>
+#include <chrono>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include <boost/uuid/uuid.hpp>
+#include <boost/uuid/uuid_generators.hpp>
+#include <boost/uuid/uuid_io.hpp>
+
+#include <folly/futures/Future.h>
+
+export module streamr.dht.connectivityChecker;
+
+import streamr.dht.protos;
+
+import streamr.dht.Connection;
+import streamr.dht.Connectivity;
+import streamr.dht.Errors;
+import streamr.dht.Version;
+import streamr.dht.WebsocketClientConnection;
+import streamr.logger.SLogger;
+
+// Hoisted from the former header (file scope, NOT exported).
+using streamr::logger::SLogger;
+
+export namespace streamr::dht::connection {
+
+using ::dht::ConnectivityRequest;
+using ::dht::ConnectivityResponse;
+using ::dht::Message;
+using ::dht::PeerDescriptor;
+using streamr::dht::connection::websocket::WebsocketClientConnection;
+using streamr::dht::helpers::ConnectionFailed;
+using streamr::dht::helpers::Connectivity;
+using streamr::dht::helpers::ConnectivityResponseTimeout;
+using streamr::dht::helpers::Version;
+
+inline constexpr auto CONNECTIVITY_CHECKER_SERVICE_ID =
+    "system/connectivity-checker";
+inline constexpr std::chrono::milliseconds connectivityCheckerTimeout{5000};
+inline constexpr std::chrono::milliseconds connectDefaultTimeout{1000};
+
+// TS connectAsync: open a websocket client connection to `url` and wait for
+// it to be connected; throws ConnectionFailed on error or timeout. The
+// returned connection is open; the caller closes it.
+inline std::shared_ptr<WebsocketClientConnection> connectSync(
+    const std::string& url,
+    bool allowSelfSignedCertificate,
+    std::chrono::milliseconds timeout = connectDefaultTimeout) {
+    auto socket = WebsocketClientConnection::newInstance();
+    // One shared promise raced by the connected and error listeners; the
+    // flag makes the loser's later firing a no-op.
+    auto promise = std::make_shared<folly::Promise<bool>>();
+    auto settled = std::make_shared<std::atomic_flag>();
+    socket->once<connectionevents::Connected>([promise, settled]() {
+        if (!settled->test_and_set()) {
+            promise->setValue(true);
+        }
+    });
+    socket->once<connectionevents::Error>(
+        [promise, settled](const std::string& /*name*/) {
+            if (!settled->test_and_set()) {
+                promise->setValue(false);
+            }
+        });
+    auto future = promise->getFuture();
+    socket->connect(url, allowSelfSignedCertificate);
+    bool connected = false;
+    try {
+        connected = std::move(future).get(timeout);
+    } catch (const folly::FutureTimeout&) {
+        settled->test_and_set(); // silence late listeners
+        socket->destroy();
+        throw ConnectionFailed("WebSocket connection timed out");
+    }
+    if (!connected) {
+        socket->destroy();
+        throw ConnectionFailed("Could not open WebSocket connection");
+    }
+    return socket;
+}
+
+// TS sendConnectivityRequest: run the connectivity check against one entry
+// point and return its ConnectivityResponse.
+inline ConnectivityResponse sendConnectivityRequest(
+    const ConnectivityRequest& request, const PeerDescriptor& entryPoint) {
+    const auto url = Connectivity::connectivityMethodToWebsocketUrl(
+        entryPoint.websocket(), "connectivityRequest");
+    SLogger::debug("Attempting connectivity check with entrypoint " + url);
+    std::shared_ptr<WebsocketClientConnection> outgoingConnection;
+    try {
+        outgoingConnection =
+            connectSync(url, request.allowselfsignedcertificate());
+    } catch (const std::exception& e) {
+        throw ConnectionFailed(
+            "Failed to connect to entrypoint for connectivity check: " + url,
+            std::string(e.what()));
+    }
+    // Register the response listener BEFORE sending the request.
+    auto promise = std::make_shared<folly::Promise<ConnectivityResponse>>();
+    auto settled = std::make_shared<std::atomic_flag>();
+    outgoingConnection->on<connectionevents::Data>(
+        [promise, settled](const std::vector<std::byte>& bytes) {
+            Message message;
+            if (!message.ParseFromArray(
+                    bytes.data(), static_cast<int>(bytes.size()))) {
+                SLogger::trace("Could not parse message");
+                return;
+            }
+            if (message.has_connectivityresponse() &&
+                !settled->test_and_set()) {
+                promise->setValue(message.connectivityresponse());
+            }
+        });
+    Message msg;
+    msg.set_serviceid(CONNECTIVITY_CHECKER_SERVICE_ID);
+    msg.set_messageid(
+        boost::uuids::to_string(boost::uuids::random_generator()()));
+    *msg.mutable_connectivityrequest() = request;
+    const auto serialized = msg.SerializeAsString();
+    std::vector<std::byte> data(serialized.size());
+    std::memcpy(data.data(), serialized.data(), serialized.size());
+    auto future = promise->getFuture();
+    outgoingConnection->send(data);
+    SLogger::trace("ConnectivityRequest sent");
+    ConnectivityResponse response;
+    try {
+        response = std::move(future).get(connectivityCheckerTimeout);
+    } catch (const folly::FutureTimeout&) {
+        settled->test_and_set();
+        outgoingConnection->close(false);
+        throw ConnectivityResponseTimeout("timeout");
+    }
+    outgoingConnection->close(false);
+    if (!Version::isMaybeSupportedVersion(response.protocolversion())) {
+        throw ConnectionFailed(
+            "Unsupported version: " + response.protocolversion());
+    }
+    return response;
+}
+
+} // namespace streamr::dht::connection