From a8415e3497b869c69f2245ae98b404571fe0d5c3 Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Fri, 10 Jul 2026 13:30:57 +0300 Subject: [PATCH 1/6] Phase B1 (part 1): connectivity checker, request handler, signed peer descriptors MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ports the first half of milestone B1 from v103.8.0-rc.3: - createPeerDescriptorSignaturePayload + createPeerDescriptor: the local node's signed peer descriptor built from a ConnectivityResponse; the node id derives from the reported ip (last 13 bytes of keccak(ip) + last 7 bytes of sign(ip)) and the descriptor is signed over the payload — both through SigningUtils' Ethereum-magic keccak/secp256k1, matching the TS EcdsaSecp256k1Evm defaults bit-exactly (wire-visible). Like the TS TODO, the key pair is throwaway (random private key, 20 random publicKey bytes). - connectivityChecker (client side): connectSync opens a raw websocket client connection with a connect timeout; sendConnectivityRequest runs the connectivity check against an entry point and returns its ConnectivityResponse (5 s response timeout, protocol-version check). Deliberately synchronous-blocking: both call sites run on dedicated worker threads, and the listeners are registered before connect()/send() so a fast completion cannot be lost. - connectivityRequestHandler (server side): parses ConnectivityRequests off a connection's data events, probes the requester's advertised websocket server (action=connectivityProbe) unless the port is 0, and replies with a ConnectivityResponse. Runs on a dedicated magic-static executor so the probe never blocks the websocket dispatch thread. GeoIP lookup omitted (deferred to milestone E). Template over the connection type so the unit test drives it with a mock (the TS test uses a bare EventEmitter). - New error types ConnectionFailed / ConnectivityResponseTimeout. Tests ported (all green, unit suite 177/177): - unit/createPeerDescriptor.test.ts -> CreatePeerDescriptorTest (3 tests) - unit/connectivityRequestHandler.test.ts -> ConnectivityRequestHandlerTest (2 tests; the happy path exercises a real websocket probe round trip against a local WebsocketServer) Also records in the plan the owner's manual follow-up: compare the KademliaCorrectness statistics against the TypeScript implementation (the TS benchmark is bit-rotted at the pin; C++ numbers and the diagnosis are in the note). Still to come in B1 (part 2): wiring the handler into WebsocketServerConnector's connectivityRequest/connectivityProbe actions, the real client-side checkConnectivity flow in the connector facade (honoring externalIp/websocketHost/port-range options), and the ConnectivityChecking/Websocket/WebsocketConnectionManagement integration tests. Co-Authored-By: Claude Fable 5 --- packages/streamr-dht/CMakeLists.txt | 2 + .../connection/connectivityChecker.cppm | 160 ++++++++++++++++ .../connectivityRequestHandler.cppm | 172 ++++++++++++++++++ .../streamr-dht/modules/helpers/Errors.cppm | 22 ++- .../modules/helpers/createPeerDescriptor.cppm | 110 +++++++++++ .../createPeerDescriptorSignaturePayload.cppm | 61 +++++++ .../unit/ConnectivityRequestHandlerTest.cpp | 171 +++++++++++++++++ .../test/unit/CreatePeerDescriptorTest.cpp | 85 +++++++++ trackerless-network-completion-plan.md | 14 ++ 9 files changed, 796 insertions(+), 1 deletion(-) create mode 100644 packages/streamr-dht/modules/connection/connectivityChecker.cppm create mode 100644 packages/streamr-dht/modules/connection/connectivityRequestHandler.cppm create mode 100644 packages/streamr-dht/modules/helpers/createPeerDescriptor.cppm create mode 100644 packages/streamr-dht/modules/helpers/createPeerDescriptorSignaturePayload.cppm create mode 100644 packages/streamr-dht/test/unit/ConnectivityRequestHandlerTest.cpp create mode 100644 packages/streamr-dht/test/unit/CreatePeerDescriptorTest.cpp diff --git a/packages/streamr-dht/CMakeLists.txt b/packages/streamr-dht/CMakeLists.txt index 092eb266..c6830e11 100644 --- a/packages/streamr-dht/CMakeLists.txt +++ b/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 diff --git a/packages/streamr-dht/modules/connection/connectivityChecker.cppm b/packages/streamr-dht/modules/connection/connectivityChecker.cppm new file mode 100644 index 00000000..d69003cd --- /dev/null +++ b/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 +#include +#include +#include +#include + +#include +#include +#include + +#include + +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 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>(); + auto settled = std::make_shared(); + socket->once([promise, settled]() { + if (!settled->test_and_set()) { + promise->setValue(true); + } + }); + socket->once( + [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 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>(); + auto settled = std::make_shared(); + outgoingConnection->on( + [promise, settled](const std::vector& bytes) { + Message message; + if (!message.ParseFromArray( + bytes.data(), static_cast(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 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 diff --git a/packages/streamr-dht/modules/connection/connectivityRequestHandler.cppm b/packages/streamr-dht/modules/connection/connectivityRequestHandler.cppm new file mode 100644 index 00000000..765218db --- /dev/null +++ b/packages/streamr-dht/modules/connection/connectivityRequestHandler.cppm @@ -0,0 +1,172 @@ +// Module streamr.dht.connectivityRequestHandler +// Ported from packages/dht/src/connection/connectivityRequestHandler.ts +// (v103.8.0-rc.3): the server side of the NAT connectivity check. Attached to +// an incoming websocket connection whose URL carried the connectivityRequest +// action; on a ConnectivityRequest it probes the requester's advertised +// websocket server (connect back to host:port, action=connectivityProbe) +// unless probing is disabled (port 0), and replies with a +// ConnectivityResponse. The GeoIP location lookup is omitted (deferred to +// milestone E with the rest of GeoIP). +// +// The handling runs on a dedicated worker executor (magic static, like +// AbortableTimers'): the probe blocks up to the connect timeout, which must +// not stall the websocket dispatch thread the data event fires on. +// attachConnectivityRequestHandler is a template over the connection type so +// the unit test can drive it with a mock connection (the TS test does the +// same with a bare EventEmitter); the real caller passes the +// WebsocketServerConnection. +module; + +#include +#include +#include + +#include +#include +#include + +#include + +export module streamr.dht.connectivityRequestHandler; + +import streamr.dht.protos; + +import streamr.dht.Connection; +import streamr.dht.Connectivity; +import streamr.dht.connectivityChecker; +import streamr.dht.NatType; +import streamr.dht.Version; +import streamr.logger.SLogger; +import streamr.utils.Ipv4Helper; + +// Hoisted from the former header (file scope, NOT exported). +using streamr::logger::SLogger; +using streamr::utils::Ipv4Helper; + +export namespace streamr::dht::connection { + +using ::dht::ConnectivityRequest; +using ::dht::ConnectivityResponse; +using ::dht::Message; +using streamr::dht::helpers::Connectivity; +using streamr::dht::helpers::Version; + +inline constexpr uint32_t DISABLE_CONNECTIVITY_PROBE = 0; + +namespace connectivityrequesthandlerdetail { + +inline folly::CPUThreadPoolExecutor& getHandlerExecutor() { + // magic static + static folly::CPUThreadPoolExecutor executor{1}; + return executor; +} + +inline ConnectivityResponse connectivityProbe( + const ConnectivityRequest& connectivityRequest, + const std::string& ipAddress, + const std::string& host) { + ConnectivityResponse response; + ::dht::ConnectivityMethod wsServerInfo; + wsServerInfo.set_host(host); + wsServerInfo.set_port(connectivityRequest.port()); + wsServerInfo.set_tls(connectivityRequest.tls()); + const auto url = Connectivity::connectivityMethodToWebsocketUrl( + wsServerInfo, "connectivityProbe"); + SLogger::trace("Attempting Connectivity Check to " + url); + try { + auto outgoingConnection = + connectSync(url, connectivityRequest.allowselfsignedcertificate()); + SLogger::trace( + "Connectivity test produced positive result, communicating" + " reply to the requester " + + host + ":" + std::to_string(connectivityRequest.port())); + response.set_host(host); + response.set_nattype(types::NatType::OPEN_INTERNET); + response.mutable_websocket()->set_host(host); + response.mutable_websocket()->set_port(connectivityRequest.port()); + response.mutable_websocket()->set_tls(connectivityRequest.tls()); + response.set_ipaddress(Ipv4Helper::ipv4ToNumber(ipAddress)); + response.set_protocolversion(Version::localProtocolVersion); + outgoingConnection->close(false); + } catch (const std::exception& err) { + SLogger::debug("error " + std::string(err.what())); + response.set_host(host); + response.set_nattype(types::NatType::UNKNOWN); + response.set_ipaddress(Ipv4Helper::ipv4ToNumber(ipAddress)); + response.set_protocolversion(Version::localProtocolVersion); + } + return response; +} + +template +inline void handleIncomingConnectivityRequest( + ConnectionType& connection, + const ConnectivityRequest& connectivityRequest) { + const auto ipAddress = connection.getRemoteAddress(); + const auto host = + connectivityRequest.has_host() ? connectivityRequest.host() : ipAddress; + ConnectivityResponse connectivityResponse; + if (connectivityRequest.port() != DISABLE_CONNECTIVITY_PROBE) { + connectivityResponse = + connectivityProbe(connectivityRequest, ipAddress, host); + } else { + SLogger::trace( + "ConnectivityRequest port is 0, replying without" + " connectivityProbe"); + connectivityResponse.set_host(host); + connectivityResponse.set_nattype(types::NatType::UNKNOWN); + connectivityResponse.set_ipaddress(Ipv4Helper::ipv4ToNumber(ipAddress)); + connectivityResponse.set_protocolversion(Version::localProtocolVersion); + } + Message msg; + msg.set_serviceid(CONNECTIVITY_CHECKER_SERVICE_ID); + msg.set_messageid( + boost::uuids::to_string(boost::uuids::random_generator()())); + *msg.mutable_connectivityresponse() = connectivityResponse; + const auto serialized = msg.SerializeAsString(); + std::vector data(serialized.size()); + std::memcpy(data.data(), serialized.data(), serialized.size()); + connection.send(data); + SLogger::trace("ConnectivityResponse sent"); +} + +} // namespace connectivityrequesthandlerdetail + +template +inline void attachConnectivityRequestHandler( + const std::shared_ptr& connectionToListenTo) { + namespace detail = connectivityrequesthandlerdetail; + std::weak_ptr weakConnection = connectionToListenTo; + connectionToListenTo->template on( + [weakConnection](const std::vector& data) { + SLogger::trace("server received data"); + Message message; + if (!message.ParseFromArray( + data.data(), static_cast(data.size()))) { + SLogger::trace("Could not parse message"); + return; + } + if (!message.has_connectivityrequest()) { + return; + } + SLogger::trace("ConnectivityRequest received"); + detail::getHandlerExecutor().add( + [weakConnection, request = message.connectivityrequest()]() { + const auto connection = weakConnection.lock(); + if (!connection) { + return; + } + try { + detail::handleIncomingConnectivityRequest( + *connection, request); + SLogger::trace("handleIncomingConnectivityRequest ok"); + } catch (const std::exception& err) { + SLogger::error( + "handleIncomingConnectivityRequest " + + std::string(err.what())); + } + }); + }); +} + +} // namespace streamr::dht::connection diff --git a/packages/streamr-dht/modules/helpers/Errors.cppm b/packages/streamr-dht/modules/helpers/Errors.cppm index b796793c..4beb8b9d 100644 --- a/packages/streamr-dht/modules/helpers/Errors.cppm +++ b/packages/streamr-dht/modules/helpers/Errors.cppm @@ -17,7 +17,9 @@ enum class ErrorCode { STARTING_WEBSOCKET_SERVER_FAILED, COULD_NOT_START, CANNOT_CONNECT_TO_SELF, - SEND_FAILED + SEND_FAILED, + CONNECTION_FAILED, + CONNECTIVITY_RESPONSE_TIMEOUT }; // NOLINTEND @@ -59,6 +61,24 @@ private: } }; +// Ported from the TS Err.ConnectionFailed / Err.ConnectivityResponseTimeout +// (phase B1, connectivityChecker). +struct ConnectionFailed : public Err { + explicit ConnectionFailed( + const std::string& message, + const std::optional& originalErrorInfo = std::nullopt) + : Err(ErrorCode::CONNECTION_FAILED, message, originalErrorInfo) {} +}; + +struct ConnectivityResponseTimeout : public Err { + explicit ConnectivityResponseTimeout( + const std::string& message, + const std::optional& originalErrorInfo = std::nullopt) + : Err(ErrorCode::CONNECTIVITY_RESPONSE_TIMEOUT, + message, + originalErrorInfo) {} +}; + struct WebsocketServerStartError : public Err { explicit WebsocketServerStartError( const std::string& message, diff --git a/packages/streamr-dht/modules/helpers/createPeerDescriptor.cppm b/packages/streamr-dht/modules/helpers/createPeerDescriptor.cppm new file mode 100644 index 00000000..6acf4b9d --- /dev/null +++ b/packages/streamr-dht/modules/helpers/createPeerDescriptor.cppm @@ -0,0 +1,110 @@ +// Module streamr.dht.createPeerDescriptor +// Ported from packages/dht/src/helpers/createPeerDescriptor.ts +// (v103.8.0-rc.3). Builds the local node's signed peer descriptor from a +// connectivity response: unless an explicit node id is given, the id is +// derived from the reported ip address as +// concat(last 13 bytes of keccak(ip), last 7 bytes of sign(ip)) +// and the whole descriptor is signed over +// createPeerDescriptorSignaturePayload. The keccak and the signature both +// use the Ethereum message magic (the TS EcdsaSecp256k1Evm defaults), which +// SigningUtils::hash / createSignature implement — wire-visible, keep exact. +// Like TS (its TODO), the key pair is throwaway: a random 32-byte private +// key signs, and the published publicKey is 20 random bytes. +module; + +#include + +#include +#include +#include +#include + +export module streamr.dht.createPeerDescriptor; + +import streamr.dht.protos; + +import streamr.dht.createPeerDescriptorSignaturePayload; +import streamr.dht.Identifiers; +import streamr.utils.BinaryUtils; +import streamr.utils.SigningUtils; + +export namespace streamr::dht::helpers { + +using ::dht::ConnectivityResponse; +using ::dht::NodeType; +using ::dht::PeerDescriptor; +using streamr::dht::DhtAddress; +using streamr::dht::DhtAddressRaw; +using streamr::dht::Identifiers; +using streamr::utils::BinaryUtils; +using streamr::utils::SigningUtils; + +namespace createpeerdescriptordetail { + +inline constexpr size_t privateKeyBytes = 32; +inline constexpr size_t publicKeyBytes = 20; +inline constexpr size_t nodeIdHashBytes = 13; +inline constexpr size_t nodeIdSignatureBytes = 7; + +inline std::string randomBytes(size_t count) { + std::string bytes(count, '\0'); + if (RAND_bytes( + reinterpret_cast(bytes.data()), // NOLINT + static_cast(count)) != 1) { + throw std::runtime_error("createPeerDescriptor: RAND_bytes failed"); + } + return bytes; +} + +// nodeId is calculated as +// concatenate( +// get104leastSignificantBits(hash(ipAddress)), +// get56leastSignificantBits(sign(ipAddress)) +// ) +inline DhtAddressRaw calculateNodeIdRaw( + uint32_t ipAddress, const std::string& privateKeyHex) { + const std::string ipAsBuffer = convertUnsignedIntegerToBuffer(ipAddress); + const std::string ipHash = SigningUtils::hash(ipAsBuffer); + const std::string signature = + SigningUtils::createSignature(ipAsBuffer, privateKeyHex); + return DhtAddressRaw{ + ipHash.substr(ipHash.size() - nodeIdHashBytes) + + signature.substr(signature.size() - nodeIdSignatureBytes)}; +} + +} // namespace createpeerdescriptordetail + +inline PeerDescriptor createPeerDescriptor( + const ConnectivityResponse& connectivityResponse, + uint32_t region, + const std::optional& nodeId = std::nullopt) { + namespace detail = createpeerdescriptordetail; + const std::string privateKeyHex = BinaryUtils::binaryStringToHex( + detail::randomBytes(detail::privateKeyBytes)); + // TODO calculate publicKey from privateKey (mirrors the TS TODO) + const std::string publicKey = detail::randomBytes(detail::publicKeyBytes); + DhtAddressRaw nodeIdRaw = (nodeId.has_value()) + ? Identifiers::getRawFromDhtAddress(*nodeId) + : detail::calculateNodeIdRaw( + connectivityResponse.ipaddress(), privateKeyHex); + PeerDescriptor ret; + ret.set_nodeid(nodeIdRaw); + ret.set_type(NodeType::NODEJS); + ret.set_ipaddress(connectivityResponse.ipaddress()); + ret.set_region(region); + ret.set_publickey(publicKey); + if (connectivityResponse.has_websocket()) { + ret.mutable_websocket()->set_host( + connectivityResponse.websocket().host()); + ret.mutable_websocket()->set_port( + connectivityResponse.websocket().port()); + ret.mutable_websocket()->set_tls( + connectivityResponse.websocket().tls()); + } + ret.set_signature( + SigningUtils::createSignature( + createPeerDescriptorSignaturePayload(ret), privateKeyHex)); + return ret; +} + +} // namespace streamr::dht::helpers diff --git a/packages/streamr-dht/modules/helpers/createPeerDescriptorSignaturePayload.cppm b/packages/streamr-dht/modules/helpers/createPeerDescriptorSignaturePayload.cppm new file mode 100644 index 00000000..08deef92 --- /dev/null +++ b/packages/streamr-dht/modules/helpers/createPeerDescriptorSignaturePayload.cppm @@ -0,0 +1,61 @@ +// Module streamr.dht.createPeerDescriptorSignaturePayload +// Ported from packages/dht/src/helpers/createPeerDescriptorSignaturePayload.ts +// (v103.8.0-rc.3). Builds the byte payload a peer descriptor's signature is +// computed over: the ','-separated concatenation of the descriptor's type, +// udp/tcp/websocket connectivity methods (protobuf-serialized), region, +// ipAddress and publicKey. Wire-visible — must stay bit-exact with TS +// (protobuf-ts toBinary == protobuf SerializeAsString for the same message). +module; + +#include +#include + +export module streamr.dht.createPeerDescriptorSignaturePayload; + +import streamr.dht.protos; + +export namespace streamr::dht::helpers { + +using ::dht::PeerDescriptor; + +// Buffer.alloc(4).writeUInt32BE(number) +inline std::string convertUnsignedIntegerToBuffer(uint32_t number) { + std::string buffer(4, '\0'); + buffer[0] = static_cast((number >> 24U) & 0xFFU); // NOLINT + buffer[1] = static_cast((number >> 16U) & 0xFFU); // NOLINT + buffer[2] = static_cast((number >> 8U) & 0xFFU); // NOLINT + buffer[3] = static_cast(number & 0xFFU); // NOLINT + return buffer; +} + +inline std::string createPeerDescriptorSignaturePayload( + const PeerDescriptor& peerDescriptor) { + // The TS scalar fields (type, region, ipAddress, publicKey) are always + // defined on a generated protobuf-ts object, so they are always included; + // the message fields (udp, tcp, websocket) only when present. + const std::string separator = ","; + std::string payload; + payload += convertUnsignedIntegerToBuffer( + static_cast(peerDescriptor.type())); + payload += separator; + if (peerDescriptor.has_udp()) { + payload += peerDescriptor.udp().SerializeAsString(); + } + payload += separator; + if (peerDescriptor.has_tcp()) { + payload += peerDescriptor.tcp().SerializeAsString(); + } + payload += separator; + if (peerDescriptor.has_websocket()) { + payload += peerDescriptor.websocket().SerializeAsString(); + } + payload += separator; + payload += convertUnsignedIntegerToBuffer(peerDescriptor.region()); + payload += separator; + payload += convertUnsignedIntegerToBuffer(peerDescriptor.ipaddress()); + payload += separator; + payload += peerDescriptor.publickey(); + return payload; +} + +} // namespace streamr::dht::helpers diff --git a/packages/streamr-dht/test/unit/ConnectivityRequestHandlerTest.cpp b/packages/streamr-dht/test/unit/ConnectivityRequestHandlerTest.cpp new file mode 100644 index 00000000..ad9f022b --- /dev/null +++ b/packages/streamr-dht/test/unit/ConnectivityRequestHandlerTest.cpp @@ -0,0 +1,171 @@ +// Ported from packages/dht/test/unit/connectivityRequestHandler.test.ts +// (v103.8.0-rc.3): the handler is attached to a mock connection; a real +// websocket server on 127.0.0.1:15001 serves as the probe target. The happy +// path replies OPEN_INTERNET with the probed websocket info; with probing +// disabled (port 0) it replies UNKNOWN without websocket info. +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include // IWYU pragma: keep + +import streamr.eventemitter.EventEmitter; +import streamr.utils.CoroutineHelper; +import streamr.utils.Ipv4Helper; +import streamr.utils.waitForCondition; +import streamr.dht.Connection; +import streamr.dht.connectivityChecker; +import streamr.dht.connectivityRequestHandler; +import streamr.dht.NatType; +import streamr.dht.Version; +import streamr.dht.WebsocketServer; +import streamr.dht.protos; + +using ::dht::ConnectivityRequest; +using ::dht::Message; +using streamr::dht::connection::attachConnectivityRequestHandler; +using streamr::dht::connection::CONNECTIVITY_CHECKER_SERVICE_ID; +using streamr::dht::connection::connectionevents::Data; +using streamr::dht::connection::websocket::WebsocketServer; +using streamr::dht::connection::websocket::WebsocketServerConfig; +using streamr::dht::helpers::Version; +using streamr::utils::blockingWait; +using streamr::utils::Ipv4Helper; +using streamr::utils::waitForCondition; +using namespace std::chrono_literals; + +namespace { + +constexpr auto host = "127.0.0.1"; +constexpr uint16_t port = 15001; +constexpr auto conditionTimeout = std::chrono::seconds(5); +constexpr auto conditionPollInterval = std::chrono::milliseconds(100); + +// The TS test drives the handler with a bare EventEmitter carrying send() +// and getRemoteIpAddress(); this mock is its C++ equivalent (the handler is +// a template over the connection type). +class MockConnection : public streamr::eventemitter::EventEmitter< + streamr::dht::connection::ConnectionEvents> { +private: + std::mutex mutex; + std::vector> sentData; + +public: + void send(const std::vector& data) { + std::scoped_lock lock(this->mutex); + this->sentData.push_back(data); + } + + [[nodiscard]] std::string getRemoteAddress() const { // NOLINT + return host; + } + + [[nodiscard]] size_t getSentCount() { + std::scoped_lock lock(this->mutex); + return this->sentData.size(); + } + + [[nodiscard]] std::vector getSent(size_t index) { + std::scoped_lock lock(this->mutex); + return this->sentData.at(index); + } +}; + +std::vector serializeMessage(const Message& message) { + const auto serialized = message.SerializeAsString(); + std::vector data(serialized.size()); + std::memcpy(data.data(), serialized.data(), serialized.size()); + return data; +} + +Message parseMessage(const std::vector& data) { + Message message; + message.ParseFromArray(data.data(), static_cast(data.size())); + return message; +} + +} // namespace + +class ConnectivityRequestHandlerTest : public ::testing::Test { +protected: + std::optional websocketServer; + std::shared_ptr connection; + + void SetUp() override { + this->websocketServer.emplace( + WebsocketServerConfig{ + .portRange = {port, port}, + .enableTls = false, + .tlsCertificateFiles = std::nullopt, + .maxMessageSize = std::nullopt}); + this->websocketServer->start(); + this->connection = std::make_shared(); + } + + void TearDown() override { this->websocketServer->stop(); } + + [[nodiscard]] static Message createRequest(uint16_t requestPort) { + Message request; + request.set_serviceid(CONNECTIVITY_CHECKER_SERVICE_ID); + request.set_messageid("mock-message-id"); + ConnectivityRequest* connectivityRequest = + request.mutable_connectivityrequest(); + connectivityRequest->set_port(requestPort); + connectivityRequest->set_host(host); + connectivityRequest->set_tls(false); + connectivityRequest->set_allowselfsignedcertificate(false); + return request; + } +}; + +TEST_F(ConnectivityRequestHandlerTest, HappyPath) { + attachConnectivityRequestHandler(this->connection); + this->connection->emit(serializeMessage(createRequest(port))); + + auto* rawConnection = this->connection.get(); + blockingWait(waitForCondition( + [rawConnection]() { return rawConnection->getSentCount() > 0; }, + conditionTimeout, + conditionPollInterval)); + + const auto receivedMessage = parseMessage(this->connection->getSent(0)); + EXPECT_EQ(receivedMessage.serviceid(), CONNECTIVITY_CHECKER_SERVICE_ID); + EXPECT_FALSE(receivedMessage.messageid().empty()); + ASSERT_TRUE(receivedMessage.has_connectivityresponse()); + const auto& response = receivedMessage.connectivityresponse(); + EXPECT_EQ(response.host(), host); + EXPECT_EQ(response.nattype(), streamr::dht::types::NatType::OPEN_INTERNET); + ASSERT_TRUE(response.has_websocket()); + EXPECT_EQ(response.websocket().host(), host); + EXPECT_EQ(response.websocket().port(), port); + EXPECT_FALSE(response.websocket().tls()); + EXPECT_EQ(response.ipaddress(), Ipv4Helper::ipv4ToNumber(host)); + EXPECT_EQ(response.protocolversion(), Version::localProtocolVersion); +} + +TEST_F(ConnectivityRequestHandlerTest, DisabledConnectivityProbing) { + attachConnectivityRequestHandler(this->connection); + this->connection->emit(serializeMessage(createRequest(0))); + + auto* rawConnection = this->connection.get(); + blockingWait(waitForCondition( + [rawConnection]() { return rawConnection->getSentCount() > 0; }, + conditionTimeout, + conditionPollInterval)); + + const auto receivedMessage = parseMessage(this->connection->getSent(0)); + EXPECT_EQ(receivedMessage.serviceid(), CONNECTIVITY_CHECKER_SERVICE_ID); + ASSERT_TRUE(receivedMessage.has_connectivityresponse()); + const auto& response = receivedMessage.connectivityresponse(); + EXPECT_EQ(response.host(), host); + EXPECT_EQ(response.nattype(), streamr::dht::types::NatType::UNKNOWN); + EXPECT_FALSE(response.has_websocket()); + EXPECT_EQ(response.ipaddress(), Ipv4Helper::ipv4ToNumber(host)); + EXPECT_EQ(response.protocolversion(), Version::localProtocolVersion); +} diff --git a/packages/streamr-dht/test/unit/CreatePeerDescriptorTest.cpp b/packages/streamr-dht/test/unit/CreatePeerDescriptorTest.cpp new file mode 100644 index 00000000..6e778a65 --- /dev/null +++ b/packages/streamr-dht/test/unit/CreatePeerDescriptorTest.cpp @@ -0,0 +1,85 @@ +// Ported from packages/dht/test/unit/createPeerDescriptor.test.ts +// (v103.8.0-rc.3). The TS expect.any(Uint8Array) assertions become size +// checks (the node id is always 20 bytes, the throwaway public key 20 bytes +// and the recoverable secp256k1 signature 65 bytes); there is no browser +// environment, so the type is always NODEJS. +#include +#include +#include + +import streamr.dht.createPeerDescriptor; +import streamr.dht.Identifiers; +import streamr.dht.RegionPings; +import streamr.dht.protos; +import streamr.utils.Ipv4Helper; + +using ::dht::ConnectivityResponse; +using ::dht::NodeType; +using streamr::dht::Identifiers; +using streamr::dht::connection::simulator::getRandomRegion; +using streamr::dht::helpers::createPeerDescriptor; +using streamr::utils::Ipv4Helper; + +namespace { + +constexpr auto ipAddressString = "1.2.3.4"; +constexpr size_t nodeIdBytes = 20; +constexpr size_t publicKeyBytes = 20; +constexpr size_t signatureBytes = 65; + +} // namespace + +class CreatePeerDescriptorTest : public ::testing::Test { +protected: + uint32_t region = static_cast(getRandomRegion()); + uint32_t ipAddress = Ipv4Helper::ipv4ToNumber(ipAddressString); +}; + +TEST_F(CreatePeerDescriptorTest, WithoutWebsocket) { + ConnectivityResponse connectivityResponse; + connectivityResponse.set_ipaddress(this->ipAddress); + const auto peerDescriptor = + createPeerDescriptor(connectivityResponse, this->region); + EXPECT_EQ(peerDescriptor.nodeid().size(), nodeIdBytes); + EXPECT_EQ(peerDescriptor.type(), NodeType::NODEJS); + EXPECT_EQ(peerDescriptor.ipaddress(), this->ipAddress); + EXPECT_EQ(peerDescriptor.publickey().size(), publicKeyBytes); + EXPECT_EQ(peerDescriptor.signature().size(), signatureBytes); + EXPECT_EQ(peerDescriptor.region(), this->region); + EXPECT_FALSE(peerDescriptor.has_websocket()); +} + +TEST_F(CreatePeerDescriptorTest, WithWebsocket) { + ConnectivityResponse connectivityResponse; + connectivityResponse.set_ipaddress(this->ipAddress); + connectivityResponse.mutable_websocket()->set_host("bar.com"); + connectivityResponse.mutable_websocket()->set_port(123); // NOLINT + connectivityResponse.mutable_websocket()->set_tls(true); + const auto peerDescriptor = + createPeerDescriptor(connectivityResponse, this->region); + EXPECT_EQ(peerDescriptor.nodeid().size(), nodeIdBytes); + EXPECT_EQ(peerDescriptor.type(), NodeType::NODEJS); + EXPECT_EQ(peerDescriptor.ipaddress(), this->ipAddress); + EXPECT_EQ(peerDescriptor.publickey().size(), publicKeyBytes); + EXPECT_EQ(peerDescriptor.signature().size(), signatureBytes); + ASSERT_TRUE(peerDescriptor.has_websocket()); + EXPECT_EQ(peerDescriptor.websocket().host(), "bar.com"); + EXPECT_EQ(peerDescriptor.websocket().port(), 123); + EXPECT_TRUE(peerDescriptor.websocket().tls()); + EXPECT_EQ(peerDescriptor.region(), this->region); +} + +TEST_F(CreatePeerDescriptorTest, ExplicitNodeId) { + const auto nodeId = Identifiers::createRandomDhtAddress(); + ConnectivityResponse connectivityResponse; + connectivityResponse.set_ipaddress(this->ipAddress); + const auto peerDescriptor = + createPeerDescriptor(connectivityResponse, this->region, nodeId); + EXPECT_EQ( + peerDescriptor.nodeid(), Identifiers::getRawFromDhtAddress(nodeId)); + EXPECT_EQ(peerDescriptor.type(), NodeType::NODEJS); + EXPECT_EQ(peerDescriptor.ipaddress(), this->ipAddress); + EXPECT_EQ(peerDescriptor.publickey().size(), publicKeyBytes); + EXPECT_EQ(peerDescriptor.signature().size(), signatureBytes); + EXPECT_EQ(peerDescriptor.region(), this->region); +} diff --git a/trackerless-network-completion-plan.md b/trackerless-network-completion-plan.md index add814da..b7e58c22 100644 --- a/trackerless-network-completion-plan.md +++ b/trackerless-network-completion-plan.md @@ -433,6 +433,20 @@ use a layer-0 `DhtNode` as its transport, sharing one `ConnectionManager` and co `integration/Layer1-scale.test.ts`, `integration/DhtNodeExternalAPI.test.ts`, `benchmark/KademliaCorrectness.test.ts` (as a slow correctness test, not a benchmark). +*Open follow-up (owner, manual): compare KademliaCorrectness against the TypeScript +implementation.* The C++ port at 200 nodes measures **6.40/8** average prefix-exact neighbours +(the strict TS metric: matching prefix of the exact global XOR ordering, truncated at the first +mismatch) and **6.98/8** order-insensitive known-of-closest-8, i.e. ~0.6 of the gap is ordering +and ~1 true-closest peer is on average genuinely absent (sequential joins learn later-joining +closer peers only passively, and the measurement runs before any recovery cycle). Whether these +numbers match the TS implementation is unverified: the TS benchmark is bit-rotted at the pinned +commit (`prepare-kademlia-simulation` does not exist and no code writes `test/data/*.json`), and +the reference checkout's jest/ts-jest setup does not run. Compatible ground-truth data can be +generated (the format is just XOR-ordered indices; a generator script exists from the 2026-07-10 +session) — the owner will run the TS benchmark manually and compare against the C++ numbers +above. If TS scores materially higher, investigate the C++ ping/eviction and contact-ordering +paths for a port deficiency. + *Milestone A exit criterion:* a network of simulated C++ `DhtNode`s joins via entry points, routes messages, and stores/fetches data — the full dht test pyramid below the connector level is green. From ef21265a05462c2a3a4029165c667b91bca1c841 Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Fri, 10 Jul 2026 15:20:53 +0300 Subject: [PATCH 2/6] Phase B1 (part 2): real connectivity checking and websocket reverse connections MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit WebsocketServerConnector now runs the ported TS flows instead of stubs: - start() wires attachConnectivityRequestHandler into the connectivityRequest action and pins connectivityProbe sockets until the prober closes them (C++ has no GC to keep an unowned server socket alive; the pin is released by onClosed dropping all listeners). - checkConnectivity() does real connectivity checking against shuffled entry points with 2 s abortable waits between attempts and WebsocketServerStartError after the list is exhausted; the options-info branch tolerates a serverless connector (empty host/port, like the TS undefined fields). - connect()/isPossibleToFormConnection()/requestConnectionFromPeer(): the reverse-connection path — ask a serverless peer over the signaling transport to open a websocket back to this node's server; the RPC notification runs detached on a dedicated executor that destroy() joins after abort (the PeerManager::stop() pattern). DefaultConnectorFacade routes createConnection through the server connector when the client connector cannot form the connection, and passes the new entryPoints option through (externalIp and the TLS/ autocertify options stay deferred to B2/E). Fixed in passing (found by the new tests): - WebsocketClientConnectorRpcRemote::requestConnection returned the lazy notify() task whose const& parameters referenced already-destroyed locals (SEGV in Any::PackFrom on first use); it now co_awaits so the locals live in its own coroutine frame. - attachConnectivityRequestHandler captured the connection weakly, so a connectivityRequest server socket had no owner and was destroyed before the request arrived; the Data listener now holds it strongly (cycle broken by onClosed's removeAllListeners). Tests ported (v103.8.0-rc.3): integration/ConnectivityChecking.test.ts, integration/Websocket.test.ts (round-trip case added to the existing WebsocketClientServerTest), integration/ WebsocketConnectionManagement.test.ts (all five cases, including the serverless reverse-connection ones). lint.sh: three new files added to the documented clangd std-type-unification exclusion list. Co-Authored-By: Claude Fable 5 --- packages/streamr-dht/CMakeLists.txt | 2 + packages/streamr-dht/lint.sh | 10 +- .../modules/connection/ConnectorFacade.cppm | 18 +- .../connectivityRequestHandler.cppm | 12 +- .../WebsocketClientConnectorRpcRemote.cppm | 6 +- .../websocket/WebsocketServerConnector.cppm | 277 +++++++++++++++--- .../integration/ConnectivityCheckingTest.cpp | 86 ++++++ .../integration/WebsocketClientServerTest.cpp | 59 ++++ .../WebsocketConnectionManagementTest.cpp | 257 ++++++++++++++++ .../unit/ConnectivityRequestHandlerTest.cpp | 2 +- 10 files changed, 678 insertions(+), 51 deletions(-) create mode 100644 packages/streamr-dht/test/integration/ConnectivityCheckingTest.cpp create mode 100644 packages/streamr-dht/test/integration/WebsocketConnectionManagementTest.cpp diff --git a/packages/streamr-dht/CMakeLists.txt b/packages/streamr-dht/CMakeLists.txt index c6830e11..77c355b4 100644 --- a/packages/streamr-dht/CMakeLists.txt +++ b/packages/streamr-dht/CMakeLists.txt @@ -262,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 diff --git a/packages/streamr-dht/lint.sh b/packages/streamr-dht/lint.sh index 26914f36..d43fc7b1 100755 --- a/packages/streamr-dht/lint.sh +++ b/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 diff --git a/packages/streamr-dht/modules/connection/ConnectorFacade.cppm b/packages/streamr-dht/modules/connection/ConnectorFacade.cppm index e6969f6b..98481ebf 100644 --- a/packages/streamr-dht/modules/connection/ConnectorFacade.cppm +++ b/packages/streamr-dht/modules/connection/ConnectorFacade.cppm @@ -7,6 +7,8 @@ module; #include #include +#include +#include #include @@ -68,7 +70,7 @@ struct DefaultConnectorFacadeOptions { Transport& transport; std::optional websocketHost = std::nullopt; std::optional websocketPortRange = std::nullopt; - // std::vector entryPoints; + std::optional> entryPoints = std::nullopt; // std::vector 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; } diff --git a/packages/streamr-dht/modules/connection/connectivityRequestHandler.cppm b/packages/streamr-dht/modules/connection/connectivityRequestHandler.cppm index 765218db..1871d33e 100644 --- a/packages/streamr-dht/modules/connection/connectivityRequestHandler.cppm +++ b/packages/streamr-dht/modules/connection/connectivityRequestHandler.cppm @@ -136,9 +136,17 @@ template inline void attachConnectivityRequestHandler( const std::shared_ptr& connectionToListenTo) { namespace detail = connectivityrequesthandlerdetail; - std::weak_ptr weakConnection = connectionToListenTo; + // The Data listener captures the connection STRONGLY: nothing else owns + // a server socket that arrived with the connectivityRequest action (the + // handshake path stores its socket in a handshaker; TS relies on GC). + // The cycle connection->listener->connection is broken when the remote + // side closes: WebsocketConnection::onClosed() removes all listeners. + // The executor task below still captures WEAKLY, so a connection that + // closes while the request waits for the worker is simply dropped. connectionToListenTo->template on( - [weakConnection](const std::vector& data) { + [strongConnection = + connectionToListenTo](const std::vector& data) { + std::weak_ptr weakConnection = strongConnection; SLogger::trace("server received data"); Message message; if (!message.ParseFromArray( diff --git a/packages/streamr-dht/modules/connection/websocket/WebsocketClientConnectorRpcRemote.cppm b/packages/streamr-dht/modules/connection/websocket/WebsocketClientConnectorRpcRemote.cppm index 5c25b746..e7db16ad 100644 --- a/packages/streamr-dht/modules/connection/websocket/WebsocketClientConnectorRpcRemote.cppm +++ b/packages/streamr-dht/modules/connection/websocket/WebsocketClientConnectorRpcRemote.cppm @@ -60,7 +60,11 @@ public: Identifiers::getNodeIdFromPeerDescriptor(getLocalPeerDescriptor())); WebsocketConnectionRequest request{}; auto options = this->formDhtRpcOptions(); - return this->getClient().requestConnection( + // co_await (not `return`): the client's notify() is a LAZY coroutine + // that holds request/options by reference — a plain return would + // destroy these locals before the task first runs (SEGV in + // Any::PackFrom); the co_await keeps them alive in this frame. + co_await this->getClient().requestConnection( std::move(request), std::move(options), this->getTimeout()); } }; diff --git a/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm b/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm index 4009d9f2..6bc94697 100644 --- a/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm +++ b/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm @@ -3,8 +3,17 @@ // streamr-dht/connection/websocket/WebsocketServerConnector.hpp // (MODERNIZATION.md Phase 2.6): this file is now the source of truth. module; +#include +#include +// std::coroutine_traits must be visible in every translation unit +// that defines OR instantiates a coroutine; it cannot arrive through +// an imported BMI. +#include // IWYU pragma: keep + #include #include +#include +#include #include #include @@ -13,13 +22,22 @@ module; #include +#include + export module streamr.dht.WebsocketServerConnector; import streamr.dht.protos; import streamr.dht.WebsocketServerConnection; +import streamr.dht.Connection; +import streamr.dht.Connectivity; +import streamr.dht.WebsocketClientConnectorRpcRemote; +import streamr.dht.connectivityChecker; +import streamr.dht.connectivityRequestHandler; +import streamr.dht.Errors; import streamr.logger.SLogger; import streamr.utils.AbortController; +import streamr.utils.CoroutineHelper; import streamr.utils.Ipv4Helper; import streamr.utils.Uuid; import streamr.dht.Handshaker; @@ -46,13 +64,18 @@ using streamr::utils::Uuid; export namespace streamr::dht::connection::websocket { +using ::dht::ConnectivityRequest; using ::dht::ConnectivityResponse; using streamr::dht::connection::IncomingHandshaker; using streamr::dht::connection::IPendingConnection; using streamr::dht::connection::PendingConnection; +using streamr::dht::connection::websocket::WebsocketClientConnectorRpcClient; +using streamr::dht::connection::websocket::WebsocketClientConnectorRpcRemote; using streamr::dht::connection::websocket::WebsocketServer; using streamr::dht::connection::websocket::WebsocketServerConnection; +using streamr::dht::helpers::Connectivity; using streamr::dht::helpers::Version; +using streamr::dht::helpers::WebsocketServerStartError; using streamr::dht::transport::ListeningRpcCommunicator; using streamr::dht::transport::Transport; using streamr::dht::types::PortRange; @@ -78,6 +101,9 @@ struct WebsocketServerConnectorOptions { std::optional geoIpDatabaseFolder = std::nullopt; }; +inline constexpr std::chrono::milliseconds entrypointRetryDelay{2000}; +inline constexpr std::chrono::milliseconds abortablePollInterval{100}; + class WebsocketServerConnector { private: WebsocketServerConnectorOptions options; @@ -90,6 +116,9 @@ private: connectingHandshakers; std::map> ongoingConnectRequests; + // Runs the detached requestConnection notifications (TS setImmediate); + // destroy() joins it after abort. + folly::CPUThreadPoolExecutor requestConnectionExecutor{1}; std::recursive_mutex mMutex; public: @@ -143,14 +172,22 @@ public: {"remoteAddress", serverSocket->getRemoteAddress()}}); if (action == "connectivityRequest") { - SLogger::trace( - "Connectivity request received, not implemented in native-sdk yet"); - // attachConnectivityRequestHandler( - // serverSocket, this.geoIpLocator) + SLogger::trace("Connectivity request received"); + // GeoIP location lookup omitted (deferred to milestone + // E with the rest of GeoIP). + attachConnectivityRequestHandler(serverSocket); } else if (action == "connectivityProbe") { - SLogger::trace( - "Connectivity probe received, not implemented in native-sdk yet"); - // no-op + // The probe only needs the websocket connection to + // open; the prober closes it (TS: no-op). The listener + // pins the socket until then (same idiom as + // attachConnectivityRequestHandler: onClosed drops all + // listeners, releasing it). + SLogger::trace("Connectivity probe received"); + serverSocket->on( + [serverSocket]( + bool /*gracefulLeave*/, + uint64_t /*code*/, + const std::string& /*reason*/) {}); } else { // The localPeerDescriptor can be undefined here as the // WS server is used for connectivity checks before the @@ -177,33 +214,124 @@ public: } } - ConnectivityResponse checkConnectivity( - bool /* allowSelfSignedCertificate */) { - std::scoped_lock lock(this->mMutex); + // TS: wait(2000, abortSignal) between entry-point attempts. + void waitAbortable(std::chrono::milliseconds duration) { + const auto deadline = std::chrono::steady_clock::now() + duration; + while (std::chrono::steady_clock::now() < deadline && + !this->abortController.getSignal().aborted) { + std::this_thread::sleep_for(abortablePollInterval); + } + } - ConnectivityResponse response; + // The state is snapshotted under the lock and the network runs without + // it: a connectivity round trip blocks for seconds and the connector + // must stay operable meanwhile (TS holds no lock either). + ConnectivityResponse checkConnectivity(bool allowSelfSignedCertificate) { + std::optional hostSnapshot; + std::optional portSnapshot; + bool serverStarted = false; + std::vector entrypoints; + { + std::scoped_lock lock(this->mMutex); + hostSnapshot = this->host; + portSnapshot = this->selectedPort; + serverStarted = (this->websocketServer != nullptr); + if (this->options.entrypoints.has_value()) { + entrypoints = this->options.entrypoints.value(); + } + } - // if already aborted or websocket server not started - if (this->abortController.getSignal().aborted || - !this->selectedPort.has_value()) { + ConnectivityResponse response; + if (this->abortController.getSignal().aborted) { response.set_host("127.0.0.1"); response.set_nattype(NatType::UNKNOWN); response.set_ipaddress(Ipv4Helper::ipv4ToNumber("127.0.0.1")); response.set_protocolversion(Version::localProtocolVersion); - } else { - // return connectivity info given in options - - response.set_host(this->host.value()); + return response; + } + if (entrypoints.empty()) { + // Return connectivity info given in options. A connector with + // no websocket server and no host reports empty host/port here + // (TS returns undefined fields); callers in that configuration + // provide their own peer descriptor and ignore the response. + response.set_host(hostSnapshot.value_or("")); response.set_nattype(NatType::OPEN_INTERNET); + // TODO: Resolve the given host name or use as is if IP was + // given (mirrors the TS TODO). response.set_ipaddress(Ipv4Helper::ipv4ToNumber("127.0.0.1")); response.set_protocolversion(Version::localProtocolVersion); - response.mutable_websocket()->set_host(this->host.value()); - response.mutable_websocket()->set_port(this->selectedPort.value()); + response.mutable_websocket()->set_host(hostSnapshot.value_or("")); + response.mutable_websocket()->set_port(portSnapshot.value_or(0)); response.mutable_websocket()->set_tls( this->options.tlsCertificateFiles.has_value()); + return response; + } + // Do real connectivity checking against the entry points, in random + // order, until one answers. + std::shuffle( + entrypoints.begin(), + entrypoints.end(), + std::mt19937{std::random_device{}()}); + for (size_t i = 0; i < entrypoints.size() && + !this->abortController.getSignal().aborted; + i++) { + const auto& entryPoint = entrypoints[i]; + try { + ConnectivityRequest connectivityRequest; + connectivityRequest.set_port( + portSnapshot.value_or(DISABLE_CONNECTIVITY_PROBE)); + if (hostSnapshot.has_value()) { + connectivityRequest.set_host(hostSnapshot.value()); + } + connectivityRequest.set_tls( + serverStarted && + this->options.serverEnableTls.value_or(false)); + connectivityRequest.set_allowselfsignedcertificate( + allowSelfSignedCertificate); + return sendConnectivityRequest(connectivityRequest, entryPoint); + } catch (const std::exception& err) { + SLogger::error( + "Failed to connect to entrypoint with id " + + Identifiers::getNodeIdFromPeerDescriptor(entryPoint) + " " + + std::string(err.what())); + this->waitAbortable(entrypointRetryDelay); + } } + std::string attemptedHosts; + for (const auto& entryPoint : entrypoints) { + if (!attemptedHosts.empty()) { + attemptedHosts += ", "; + } + attemptedHosts += entryPoint.websocket().host() + ":" + + std::to_string(entryPoint.websocket().port()); + } + throw WebsocketServerStartError( + "Failed to connect to the entrypoints after " + + std::to_string(entrypoints.size()) + + " attempts\nAttempted hosts: " + attemptedHosts); + } + + bool isPossibleToFormConnection( + const PeerDescriptor& targetPeerDescriptor) { + std::scoped_lock lock(this->mMutex); + const auto connectionType = Connectivity::expectedConnectionType( + this->localPeerDescriptor.value(), targetPeerDescriptor); + return connectionType == ConnectionType::WEBSOCKET_SERVER; + } - return response; + // Ask the target (which has no websocket server) to open a websocket + // back to this node; the returned pending connection completes when the + // incoming handshake matches ongoingConnectRequests (attachHandshaker). + std::shared_ptr connect( + const PeerDescriptor& targetPeerDescriptor) { + std::scoped_lock lock(this->mMutex); + const auto nodeId = + Identifiers::getNodeIdFromPeerDescriptor(targetPeerDescriptor); + if (this->ongoingConnectRequests.contains(nodeId)) { + return this->ongoingConnectRequests.at(nodeId); + } + return this->requestConnectionFromPeer( + this->localPeerDescriptor.value(), targetPeerDescriptor); } void setLocalPeerDescriptor(const PeerDescriptor& localPeerDescriptor) { @@ -213,38 +341,103 @@ public: void destroy() { SLogger::trace("WebsocketServerConnector::destroy() called"); - std::scoped_lock lock(this->mMutex); - this->abortController.abort(); - - SLogger::trace("Closing ongoing connect requests"); - for (const auto& request : this->ongoingConnectRequests) { - if (request.second) { - request.second->close(true); + { + std::scoped_lock lock(this->mMutex); + this->abortController.abort(); + + SLogger::trace("Closing ongoing connect requests"); + for (const auto& request : this->ongoingConnectRequests) { + if (request.second) { + request.second->close(true); + } } - } - SLogger::trace("Closing connecting handshakers"); - for (const auto& handshaker : this->connectingHandshakers) { - if (handshaker.second && - handshaker.second->getPendingConnection()) { - handshaker.second->getPendingConnection()->close(true); + SLogger::trace("Closing connecting handshakers"); + for (const auto& handshaker : this->connectingHandshakers) { + if (handshaker.second && + handshaker.second->getPendingConnection()) { + handshaker.second->getPendingConnection()->close(true); + } } - } - SLogger::trace("Clearing maps"); - this->connectingHandshakers.clear(); - this->ongoingConnectRequests.clear(); + SLogger::trace("Clearing maps"); + this->connectingHandshakers.clear(); + this->ongoingConnectRequests.clear(); - SLogger::trace("Stopping websocket server"); - if (this->websocketServer) { - this->websocketServer->stop(); - this->websocketServer.reset(); + SLogger::trace("Stopping websocket server"); + if (this->websocketServer) { + this->websocketServer->stop(); + this->websocketServer.reset(); + } } + // Drained OUTSIDE the mutex (the PeerManager::stop() pattern): a + // detached requestConnection notification still in flight references + // this object and the RPC communicator; abort above cancels it, the + // join makes destroy() wait for it without deadlocking on mMutex. + this->requestConnectionExecutor.join(); SLogger::trace("WebsocketServerConnector::destroy() completed"); } private: + // TS requestConnectionFromPeer. The caller holds mMutex. TS defers the + // notification with setImmediate; here it runs on a dedicated worker + // executor so connect() never blocks on RPC I/O. + std::shared_ptr requestConnectionFromPeer( + const PeerDescriptor& localPeerDescriptor, + const PeerDescriptor& targetPeerDescriptor) { + auto pendingConnection = + PendingConnection::newInstance(targetPeerDescriptor); + const auto nodeId = + Identifiers::getNodeIdFromPeerDescriptor(targetPeerDescriptor); + // TS delFunc: drop the ongoing request once it settles either way + // (erasing an already-erased key is a no-op). + pendingConnection->on( + [this, nodeId]( + const PeerDescriptor& /*peerDescriptor*/, + const std::shared_ptr& /*connection*/) { + std::scoped_lock lock(this->mMutex); + this->ongoingConnectRequests.erase(nodeId); + }); + pendingConnection->on( + [this, nodeId](bool /*gracefulLeave*/) { + std::scoped_lock lock(this->mMutex); + this->ongoingConnectRequests.erase(nodeId); + }); + this->ongoingConnectRequests.emplace(nodeId, pendingConnection); + streamr::utils::co_withExecutor( + &this->requestConnectionExecutor, + folly::coro::co_invoke( + [this, + localPeerDescriptor, + targetPeerDescriptor]() -> folly::coro::Task { + try { + WebsocketClientConnectorRpcClient client( + this->options.rpcCommunicator); + WebsocketClientConnectorRpcRemote remoteConnector( + PeerDescriptor(localPeerDescriptor), + PeerDescriptor(targetPeerDescriptor), + std::move(client)); + // Cancellable by destroy(): abort fires before the + // executor join drains this task. + co_await streamr::utils::co_withCancellation( + this->abortController.getSignal() + .getCancellationToken(), + remoteConnector.requestConnection()); + SLogger::trace( + "Sent WebsocketConnectionRequest notification" + " to peer"); + } catch (const std::exception& err) { + SLogger::debug( + "Failed to send WebsocketConnectionRequest" + " notification to peer " + + std::string(err.what())); + } + })) + .start(); + return pendingConnection; + } + void attachHandshaker( const std::shared_ptr& serverSocket) { std::scoped_lock lock(this->mMutex); diff --git a/packages/streamr-dht/test/integration/ConnectivityCheckingTest.cpp b/packages/streamr-dht/test/integration/ConnectivityCheckingTest.cpp new file mode 100644 index 00000000..2e3b2931 --- /dev/null +++ b/packages/streamr-dht/test/integration/ConnectivityCheckingTest.cpp @@ -0,0 +1,86 @@ +// Ported from packages/dht/test/integration/ConnectivityChecking.test.ts +// (v103.8.0-rc.3): a ConnectionManager with a real websocket server answers +// a connectivity request sent with the client-side connectivityChecker. +// Adaptation: TS backs the connector facade with a no-op MockTransport; here +// a FakeTransport in its own FakeEnvironment serves the same purpose (the +// connector RPC communicator is not exercised by this test). +#include +#include +#include "packages/dht/protos/DhtRpc.pb.h" + +import streamr.dht.ConnectionManager; +import streamr.dht.ConnectorFacade; +import streamr.dht.FakeTransport; +import streamr.dht.PortRange; +import streamr.dht.TestUtils; +import streamr.dht.Version; +import streamr.dht.connectivityChecker; +import streamr.dht.protos; + +using ::dht::ConnectivityRequest; +using ::dht::ConnectivityResponse; +using ::dht::PeerDescriptor; +using streamr::dht::connection::ConnectionManager; +using streamr::dht::connection::ConnectionManagerOptions; +using streamr::dht::connection::DefaultConnectorFacade; +using streamr::dht::connection::DefaultConnectorFacadeOptions; +using streamr::dht::connection::sendConnectivityRequest; +using streamr::dht::helpers::Version; +using streamr::dht::testutils::createMockPeerDescriptor; +using streamr::dht::transport::FakeEnvironment; +using streamr::dht::transport::FakeTransport; +using streamr::dht::types::PortRange; + +namespace { + +constexpr uint16_t serverPort = 15000; +constexpr auto serverHost = "127.0.0.1"; + +} // namespace + +class ConnectivityCheckingTest : public ::testing::Test { +protected: + FakeEnvironment fakeEnvironment; + std::shared_ptr transport; + std::shared_ptr server; + + void SetUp() override { + this->transport = + this->fakeEnvironment.createTransport(createMockPeerDescriptor()); + DefaultConnectorFacadeOptions facadeOptions{ + .transport = *this->transport, + .websocketHost = serverHost, + .websocketPortRange = + PortRange{.min = serverPort, .max = serverPort}, + .createLocalPeerDescriptor = + [](const ConnectivityResponse& /* response */) + -> PeerDescriptor { + auto descriptor = createMockPeerDescriptor(); + descriptor.mutable_websocket()->set_host(serverHost); + descriptor.mutable_websocket()->set_port(serverPort); + descriptor.mutable_websocket()->set_tls(false); + return descriptor; + }}; + this->server = + std::make_shared(ConnectionManagerOptions{ + .createConnectorFacade = [opts = std::move(facadeOptions)]() + -> std::shared_ptr { + return std::make_shared(opts); + }}); + this->server->start(); + } + + void TearDown() override { this->server->stop(); } +}; + +TEST_F(ConnectivityCheckingTest, ConnectivityCheckWithCompatibleVersion) { + ConnectivityRequest request; + request.set_host(serverHost); + request.set_port(serverPort); + request.set_tls(false); + request.set_allowselfsignedcertificate(false); + + const auto response = sendConnectivityRequest( + request, this->server->getLocalPeerDescriptor()); + EXPECT_EQ(response.protocolversion(), Version::localProtocolVersion); +} diff --git a/packages/streamr-dht/test/integration/WebsocketClientServerTest.cpp b/packages/streamr-dht/test/integration/WebsocketClientServerTest.cpp index 056ea042..0933071e 100644 --- a/packages/streamr-dht/test/integration/WebsocketClientServerTest.cpp +++ b/packages/streamr-dht/test/integration/WebsocketClientServerTest.cpp @@ -1,3 +1,5 @@ +#include +#include #include #include #include @@ -105,3 +107,60 @@ TEST(WebsocketClientServerTest, TestClientCanTrasmitMessageToServer) { serverConnection->close(false); server.stop(); } + +// Ported from packages/dht/test/integration/Websocket.test.ts +// (v103.8.0-rc.3) 'Happy path': the server sends first as soon as the +// connection is up, the client echoes the payload back, and the server +// asserts the round trip. Adaptations: the C++ connection events do not +// replay to late listeners, so each side attaches its Data listener before +// any send can race it (the TS event loop gives the same ordering for +// free), and the port is unique to this suite instead of the TS 9977. +TEST(WebsocketClientServerTest, TestServerClientRoundTripMessageExchange) { + rtc::InitLogger(rtc::LogLevel::Verbose); + constexpr uint16_t roundTripPort = 10002; + WebsocketServerConfig config{ + .portRange = {.min = roundTripPort, .max = roundTripPort}, + .enableTls = false, + .tlsCertificateFiles = std::nullopt, + .maxMessageSize = std::nullopt}; + + WebsocketServer server(std::move(config)); + + const std::vector payload{ + std::byte{1}, std::byte{2}, std::byte{3}, std::byte{4}}; + + std::promise> clientReceivedPromise; + std::promise> serverReceivedPromise; + + std::shared_ptr serverConnection; + + server.on( + [&](const std::shared_ptr& connection) { + serverConnection = connection; + serverConnection->on( + [&](const std::vector& message) { + serverReceivedPromise.set_value(message); + }); + serverConnection->send(payload); + }); + + server.start(); + + auto client = WebsocketClientConnection::newInstance(); + client->on([&](const std::vector& message) { + clientReceivedPromise.set_value(message); + // Echo the payload back to the server. + client->send(message); + }); + + client->connect("ws://127.0.0.1:" + std::to_string(roundTripPort), false); + + const auto clientReceived = clientReceivedPromise.get_future().get(); + ASSERT_EQ(clientReceived, payload); + const auto serverReceived = serverReceivedPromise.get_future().get(); + ASSERT_EQ(serverReceived, payload); + + client->close(false); + serverConnection->close(false); + server.stop(); +} diff --git a/packages/streamr-dht/test/integration/WebsocketConnectionManagementTest.cpp b/packages/streamr-dht/test/integration/WebsocketConnectionManagementTest.cpp new file mode 100644 index 00000000..2347dabf --- /dev/null +++ b/packages/streamr-dht/test/integration/WebsocketConnectionManagementTest.cpp @@ -0,0 +1,257 @@ +// Ported from packages/dht/test/integration/ +// WebsocketConnectionManagement.test.ts (v103.8.0-rc.3): three +// ConnectionManagers signal over a shared Simulator; one runs a real +// websocket server, the other two are serverless. Opening a connection +// from the server peer to a serverless peer exercises the +// WebsocketServerConnector requestConnection path (the serverless peer is +// asked over the simulator to open a websocket back to the server); +// the reverse direction exercises the WebsocketClientConnector. +// Adaptations: jest done-callbacks and waitForEvent become +// waitForCondition on captured state; the failed-request cleanup case +// polls hasConnection() to false instead of racing a 'disconnected' +// waitForEvent against the send (the TS Promise.allSettled tolerates +// either finishing first, so the final assertion is the same). +#include +#include +#include +#include +#include +#include "packages/dht/protos/DhtRpc.pb.h" + +#include // IWYU pragma: keep + +import streamr.dht.ConnectionManager; +import streamr.dht.ConnectorFacade; +import streamr.dht.Errors; +import streamr.dht.Identifiers; +import streamr.dht.PortRange; +import streamr.dht.Simulator; +import streamr.dht.SimulatorTransport; +import streamr.dht.TestUtils; +import streamr.dht.Transport; +import streamr.dht.protos; +import streamr.utils.CoroutineHelper; +import streamr.utils.waitForCondition; + +using ::dht::ConnectivityResponse; +using ::dht::Message; +using ::dht::PeerDescriptor; +using streamr::dht::DhtAddress; +using streamr::dht::Identifiers; +using streamr::dht::connection::ConnectionManager; +using streamr::dht::connection::ConnectionManagerOptions; +using streamr::dht::connection::DefaultConnectorFacade; +using streamr::dht::connection::DefaultConnectorFacadeOptions; +using streamr::dht::connection::simulator::Simulator; +using streamr::dht::connection::simulator::SimulatorTransport; +using streamr::dht::helpers::CannotConnectToSelf; +using streamr::dht::testutils::createMockPeerDescriptor; +using streamr::dht::types::PortRange; +using streamr::utils::waitForCondition; + +namespace transportevents = streamr::dht::transport::transportevents; + +namespace { + +constexpr auto SERVICE_ID = "test"; +constexpr uint16_t wsServerPort = 12223; +// The failed-request case waits out the 15 s PendingConnection connect +// watchdog (the TS test uses a 15 s waitForEvent inside a 20 s jest +// timeout). +constexpr auto failedRequestCleanupTimeout = std::chrono::seconds(20); +constexpr auto conditionPollInterval = std::chrono::milliseconds(200); + +Message createDummyMessage(const PeerDescriptor& targetDescriptor) { + Message message; + message.set_serviceid(SERVICE_ID); + message.mutable_rpcmessage(); // body.oneofKind = rpcMessage + message.set_messageid("mockerer"); + message.mutable_targetdescriptor()->CopyFrom(targetDescriptor); + return message; +} + +std::shared_ptr createConnectionManager( + DefaultConnectorFacadeOptions options) { + return std::make_shared(ConnectionManagerOptions{ + .createConnectorFacade = [opts = std::move(options)]() + -> std::shared_ptr { + return std::make_shared(opts); + }}); +} + +void expectCondition( + const char* label, + std::function&& condition, + std::chrono::milliseconds timeout = std::chrono::seconds(15)) { + SCOPED_TRACE(label); + auto task = + waitForCondition(std::move(condition), timeout, conditionPollInterval); + EXPECT_NO_THROW(streamr::utils::blockingWait(std::move(task))); +} + +} // namespace + +class WebsocketConnectionManagementTest : public ::testing::Test { +protected: + // NOLINTBEGIN(misc-non-private-member-variables-in-classes) + Simulator simulator; + PeerDescriptor wsServerConnectorPeerDescriptor; + PeerDescriptor noWsServerConnectorPeerDescriptor; + PeerDescriptor biggerNoWsServerConnectorPeerDescriptor; + std::shared_ptr connectorTransport1; + std::shared_ptr connectorTransport2; + std::shared_ptr connectorTransport3; + std::shared_ptr wsServerManager; + std::shared_ptr noWsServerManager; + std::shared_ptr biggerNoWsServerManager; + // NOLINTEND(misc-non-private-member-variables-in-classes) + + void SetUp() override { + this->wsServerConnectorPeerDescriptor = createMockPeerDescriptor(); + this->wsServerConnectorPeerDescriptor.mutable_websocket()->set_host( + "127.0.0.1"); + this->wsServerConnectorPeerDescriptor.mutable_websocket()->set_port( + wsServerPort); + this->wsServerConnectorPeerDescriptor.mutable_websocket()->set_tls( + false); + this->noWsServerConnectorPeerDescriptor = createMockPeerDescriptor(); + this->biggerNoWsServerConnectorPeerDescriptor = + createMockPeerDescriptor(); + + this->connectorTransport1 = std::make_shared( + this->wsServerConnectorPeerDescriptor, this->simulator); + this->connectorTransport1->start(); + this->connectorTransport2 = std::make_shared( + this->noWsServerConnectorPeerDescriptor, this->simulator); + this->connectorTransport2->start(); + this->connectorTransport3 = std::make_shared( + this->biggerNoWsServerConnectorPeerDescriptor, this->simulator); + this->connectorTransport3->start(); + + const auto wsServerDescriptor = this->wsServerConnectorPeerDescriptor; + this->wsServerManager = createConnectionManager( + DefaultConnectorFacadeOptions{ + .transport = *this->connectorTransport1, + .websocketHost = "127.0.0.1", + .websocketPortRange = + PortRange{.min = wsServerPort, .max = wsServerPort}, + .createLocalPeerDescriptor = + [wsServerDescriptor]( + const ConnectivityResponse& /* response */) + -> PeerDescriptor { return wsServerDescriptor; }}); + + const auto noWsServerDescriptor = + this->noWsServerConnectorPeerDescriptor; + this->noWsServerManager = createConnectionManager( + DefaultConnectorFacadeOptions{ + .transport = *this->connectorTransport2, + .createLocalPeerDescriptor = + [noWsServerDescriptor]( + const ConnectivityResponse& /* response */) + -> PeerDescriptor { return noWsServerDescriptor; }}); + + const auto biggerNoWsServerDescriptor = + this->biggerNoWsServerConnectorPeerDescriptor; + this->biggerNoWsServerManager = createConnectionManager( + DefaultConnectorFacadeOptions{ + .transport = *this->connectorTransport3, + .createLocalPeerDescriptor = + [biggerNoWsServerDescriptor]( + const ConnectivityResponse& /* response */) + -> PeerDescriptor { return biggerNoWsServerDescriptor; }}); + + this->wsServerManager->start(); + this->noWsServerManager->start(); + this->biggerNoWsServerManager->start(); + } + + void TearDown() override { + this->wsServerManager->stop(); + this->noWsServerManager->stop(); + this->biggerNoWsServerManager->stop(); + this->connectorTransport1->stop(); + this->connectorTransport2->stop(); + this->connectorTransport3->stop(); + } +}; + +TEST_F( + WebsocketConnectionManagementTest, + CanOpenConnectionsToServerlessPeerWithSmallerNodeId) { + const auto dummyMessage = + createDummyMessage(this->noWsServerConnectorPeerDescriptor); + std::atomic received = false; + this->noWsServerManager->on( + [&received](const Message& message) { + EXPECT_EQ(message.messageid(), "mockerer"); + received = true; + }); + this->wsServerManager->send(dummyMessage, {}); + expectCondition( + "message received", [&received]() { return received.load(); }); +} + +TEST_F( + WebsocketConnectionManagementTest, + CanOpenConnectionsToServerlessPeerWithBiggerNodeId) { + const auto dummyMessage = + createDummyMessage(this->biggerNoWsServerConnectorPeerDescriptor); + std::atomic received = false; + this->biggerNoWsServerManager->on( + [&received](const Message& message) { + EXPECT_EQ(message.messageid(), "mockerer"); + received = true; + }); + this->wsServerManager->send(dummyMessage, {}); + expectCondition( + "message received", [&received]() { return received.load(); }); +} + +TEST_F( + WebsocketConnectionManagementTest, FailedConnectionRequestsAreCleanedUp) { + const auto targetDescriptor = createMockPeerDescriptor(); + const auto dummyMessage = createDummyMessage(targetDescriptor); + const auto nodeId = + Identifiers::getNodeIdFromPeerDescriptor(targetDescriptor); + // The TS send() promise rejects when the pending connection fails; + // the C++ send() buffers into the connecting endpoint and returns. + this->wsServerManager->send(dummyMessage, {}); + auto* manager = this->wsServerManager.get(); + expectCondition( + "failed connection request cleaned up", + [manager, nodeId]() { return !manager->hasConnection(nodeId); }, + failedRequestCleanupTimeout); +} + +TEST_F(WebsocketConnectionManagementTest, CanOpenConnectionsToPeerWithServer) { + const auto dummyMessage = + createDummyMessage(this->wsServerConnectorPeerDescriptor); + this->noWsServerManager->send(dummyMessage, {}); + const auto noWsServerNodeId = Identifiers::getNodeIdFromPeerDescriptor( + this->noWsServerConnectorPeerDescriptor); + const auto wsServerNodeId = Identifiers::getNodeIdFromPeerDescriptor( + this->wsServerConnectorPeerDescriptor); + auto* serverManager = this->wsServerManager.get(); + auto* serverlessManager = this->noWsServerManager.get(); + expectCondition( + "server sees connection", [serverManager, noWsServerNodeId]() { + return serverManager->hasConnection(noWsServerNodeId); + }); + expectCondition( + "serverless peer sees connection", + [serverlessManager, wsServerNodeId]() { + return serverlessManager->hasConnection(wsServerNodeId); + }); +} + +TEST_F(WebsocketConnectionManagementTest, ConnectingToSelfThrows) { + const auto selfMessage1 = + createDummyMessage(this->noWsServerConnectorPeerDescriptor); + EXPECT_THROW( + this->noWsServerManager->send(selfMessage1, {}), CannotConnectToSelf); + + const auto selfMessage2 = + createDummyMessage(this->wsServerConnectorPeerDescriptor); + EXPECT_THROW( + this->wsServerManager->send(selfMessage2, {}), CannotConnectToSelf); +} diff --git a/packages/streamr-dht/test/unit/ConnectivityRequestHandlerTest.cpp b/packages/streamr-dht/test/unit/ConnectivityRequestHandlerTest.cpp index ad9f022b..582d477c 100644 --- a/packages/streamr-dht/test/unit/ConnectivityRequestHandlerTest.cpp +++ b/packages/streamr-dht/test/unit/ConnectivityRequestHandlerTest.cpp @@ -100,7 +100,7 @@ class ConnectivityRequestHandlerTest : public ::testing::Test { void SetUp() override { this->websocketServer.emplace( WebsocketServerConfig{ - .portRange = {port, port}, + .portRange = {.min = port, .max = port}, .enableTls = false, .tlsCertificateFiles = std::nullopt, .maxMessageSize = std::nullopt}); From 178d54128c81018848e15c0ae9ba43f8986300dd Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Fri, 10 Jul 2026 16:35:33 +0300 Subject: [PATCH 3/6] CI: raise ctest per-test timeout to clear the heavy DHT convergence tests The macOS CI leg has been red since PR #73 landed: ctest's --timeout 300 (sized when the whole suite ran in ~35 s) kills Layer1ScaleTest.MultipleLayer1Dht (~214 s on a fast 10-core dev machine, longer on the 3-core macOS runners) and KademliaCorrectnessTest, then fails the until-pass retry the same way. The faster Linux runners stay under the limit, which is why only macOS failed. 1200 s clears the heavy tests with headroom while still converting a genuine hang into a failure. Co-Authored-By: Claude Fable 5 --- test.sh | 23 ++++++++++++++--------- 1 file changed, 14 insertions(+), 9 deletions(-) diff --git a/test.sh b/test.sh index fd152fc1..4f48eeeb 100755 --- a/test.sh +++ b/test.sh @@ -6,17 +6,22 @@ cmake --build build || exit 1 cd build -# --repeat until-pass:2 retries a failed test once, and --timeout 300 -# converts a hung test into a failure (which then gets the retry): the -# networking integration tests (e.g. ConnectionLockingTest) are -# timing-sensitive on shared CI runners and have both flaked and hung -# there, on the old toolchain as well as the new one. A genuinely broken -# test still fails both attempts. The whole suite normally runs in ~35 s; -# 300 s per test is generous. +# --repeat until-pass:2 retries a failed test once, and --timeout converts +# a hung test into a failure (which then gets the retry): the networking +# integration tests (e.g. ConnectionLockingTest) are timing-sensitive on +# shared CI runners and have both flaked and hung there, on the old +# toolchain as well as the new one. A genuinely broken test still fails +# both attempts. +# The timeout must clear the heavy DHT convergence tests from milestone A +# (PR #73): Layer1ScaleTest.MultipleLayer1Dht runs 48 nodes x 4 layer-1 +# services and takes ~214 s on a fast 10-core dev machine — on the 3-core +# macOS CI runners it exceeded the previous 300 s limit, so ctest killed +# it (and KademliaCorrectnessTest), failed the retry the same way, and +# the macOS leg went red while the faster Linux legs stayed green. if [ "$#" -gt 0 ]; then - ctest -V --repeat until-pass:2 --timeout 300 -R "$@" + ctest -V --repeat until-pass:2 --timeout 1200 -R "$@" else - ctest -V --repeat until-pass:2 --timeout 300 + ctest -V --repeat until-pass:2 --timeout 1200 fi CTEST_RETURN_CODE=$? From 66eace2e89caa2c44f5bac8d518acb49986a5bc5 Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Fri, 10 Jul 2026 23:40:29 +0300 Subject: [PATCH 4/6] WebsocketServerConnector: move requestConnection notifications to the shared pool Applies the PR #75 executor architecture to the one per-instance pool B1 added: the detached requestConnection notifications now run on a serial view of the shared worker pool, tracked by a GuardedAsyncScope that destroy() drains after abort (outside mMutex, the PeerManager::stop() pattern); the scope's gate drops a connect() racing the drain. Co-Authored-By: Claude Fable 5 --- .../websocket/WebsocketServerConnector.cppm | 75 ++++++++++--------- 1 file changed, 40 insertions(+), 35 deletions(-) diff --git a/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm b/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm index 6bc94697..08a41c03 100644 --- a/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm +++ b/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm @@ -38,6 +38,8 @@ import streamr.dht.Errors; import streamr.logger.SLogger; import streamr.utils.AbortController; import streamr.utils.CoroutineHelper; +import streamr.utils.GuardedAsyncScope; +import streamr.utils.SharedExecutors; import streamr.utils.Ipv4Helper; import streamr.utils.Uuid; import streamr.dht.Handshaker; @@ -116,9 +118,13 @@ private: connectingHandshakers; std::map> ongoingConnectRequests; - // Runs the detached requestConnection notifications (TS setImmediate); - // destroy() joins it after abort. - folly::CPUThreadPoolExecutor requestConnectionExecutor{1}; + // Runs the detached requestConnection notifications (TS setImmediate) + // on a serial view of the shared worker pool (per-instance pools are + // forbidden — see streamr.utils.SharedExecutors); destroy() drains the + // scope after abort, and the gate drops a connect() racing the drain. + streamr::utils::SharedSerialExecutor requestConnectionExecutor{ + streamr::utils::SharedExecutors::worker()}; + streamr::utils::GuardedAsyncScope requestConnectionScope; std::recursive_mutex mMutex; public: @@ -373,8 +379,8 @@ public: // Drained OUTSIDE the mutex (the PeerManager::stop() pattern): a // detached requestConnection notification still in flight references // this object and the RPC communicator; abort above cancels it, the - // join makes destroy() wait for it without deadlocking on mMutex. - this->requestConnectionExecutor.join(); + // drain makes destroy() wait for it without deadlocking on mMutex. + this->requestConnectionScope.close(); SLogger::trace("WebsocketServerConnector::destroy() completed"); } @@ -405,36 +411,35 @@ private: this->ongoingConnectRequests.erase(nodeId); }); this->ongoingConnectRequests.emplace(nodeId, pendingConnection); - streamr::utils::co_withExecutor( - &this->requestConnectionExecutor, - folly::coro::co_invoke( - [this, - localPeerDescriptor, - targetPeerDescriptor]() -> folly::coro::Task { - try { - WebsocketClientConnectorRpcClient client( - this->options.rpcCommunicator); - WebsocketClientConnectorRpcRemote remoteConnector( - PeerDescriptor(localPeerDescriptor), - PeerDescriptor(targetPeerDescriptor), - std::move(client)); - // Cancellable by destroy(): abort fires before the - // executor join drains this task. - co_await streamr::utils::co_withCancellation( - this->abortController.getSignal() - .getCancellationToken(), - remoteConnector.requestConnection()); - SLogger::trace( - "Sent WebsocketConnectionRequest notification" - " to peer"); - } catch (const std::exception& err) { - SLogger::debug( - "Failed to send WebsocketConnectionRequest" - " notification to peer " + - std::string(err.what())); - } - })) - .start(); + this->requestConnectionScope.add( + streamr::utils::co_withExecutor( + &this->requestConnectionExecutor, + folly::coro::co_invoke( + [this, localPeerDescriptor, targetPeerDescriptor]() + -> folly::coro::Task { + try { + WebsocketClientConnectorRpcClient client( + this->options.rpcCommunicator); + WebsocketClientConnectorRpcRemote remoteConnector( + PeerDescriptor(localPeerDescriptor), + PeerDescriptor(targetPeerDescriptor), + std::move(client)); + // Cancellable by destroy(): abort fires before the + // executor join drains this task. + co_await streamr::utils::co_withCancellation( + this->abortController.getSignal() + .getCancellationToken(), + remoteConnector.requestConnection()); + SLogger::trace( + "Sent WebsocketConnectionRequest notification" + " to peer"); + } catch (const std::exception& err) { + SLogger::debug( + "Failed to send WebsocketConnectionRequest" + " notification to peer " + + std::string(err.what())); + } + }))); return pendingConnection; } From 2f5bcbe431139936669039b2f836b8db08ae3d59 Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Sat, 11 Jul 2026 00:32:38 +0300 Subject: [PATCH 5/6] DEBUG: instrument the reverse-connection notify path (temporary) The ubuntu-latest leg SIGSEGVs (address 0x20) deterministically in the three WebsocketConnectionManagement reverse-connection tests while macOS and local arm64 runs (including under ASan) are green. These INFO logs bracket every step of requestConnectionFromPeer -> GuardedAsyncScope -> ClientApi::notify so the next CI run shows the exact crash point. Will be reverted once the cause is fixed. Co-Authored-By: Claude Fable 5 --- .../connection/websocket/WebsocketServerConnector.cppm | 9 ++++++--- .../modules/RpcCommunicatorClientApi.cppm | 7 ++++++- packages/streamr-utils/modules/GuardedAsyncScope.cppm | 3 +++ 3 files changed, 15 insertions(+), 4 deletions(-) diff --git a/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm b/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm index 08a41c03..f6b220c2 100644 --- a/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm +++ b/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm @@ -411,6 +411,7 @@ private: this->ongoingConnectRequests.erase(nodeId); }); this->ongoingConnectRequests.emplace(nodeId, pendingConnection); + SLogger::info("DBG rcfp: before scope.add"); this->requestConnectionScope.add( streamr::utils::co_withExecutor( &this->requestConnectionExecutor, @@ -418,21 +419,23 @@ private: [this, localPeerDescriptor, targetPeerDescriptor]() -> folly::coro::Task { try { + SLogger::info("DBG rcfp-task: entered"); WebsocketClientConnectorRpcClient client( this->options.rpcCommunicator); + SLogger::info("DBG rcfp-task: client built"); WebsocketClientConnectorRpcRemote remoteConnector( PeerDescriptor(localPeerDescriptor), PeerDescriptor(targetPeerDescriptor), std::move(client)); + SLogger::info("DBG rcfp-task: remote built"); // Cancellable by destroy(): abort fires before the // executor join drains this task. co_await streamr::utils::co_withCancellation( this->abortController.getSignal() .getCancellationToken(), remoteConnector.requestConnection()); - SLogger::trace( - "Sent WebsocketConnectionRequest notification" - " to peer"); + SLogger::info( + "DBG rcfp-task: notify co_await done"); } catch (const std::exception& err) { SLogger::debug( "Failed to send WebsocketConnectionRequest" diff --git a/packages/streamr-proto-rpc/modules/RpcCommunicatorClientApi.cppm b/packages/streamr-proto-rpc/modules/RpcCommunicatorClientApi.cppm index c7e993fa..0068cf11 100644 --- a/packages/streamr-proto-rpc/modules/RpcCommunicatorClientApi.cppm +++ b/packages/streamr-proto-rpc/modules/RpcCommunicatorClientApi.cppm @@ -288,7 +288,7 @@ public: const RequestType& notificationParam, const CallContextType& callContext, std::optional timeout = std::nullopt) { - SLogger::trace("notify() notificationName:", notificationName); + SLogger::info("DBG notify: entered"); std::chrono::milliseconds timeoutValue = mRpcRequestTimeout; if (timeout.has_value()) { @@ -300,9 +300,11 @@ public: notificationName); auto requestMessage = this->createRequestRpcMessage( notificationName, notificationParam, true); + SLogger::info("DBG notify: message created"); auto&& promiseContract = folly::coro::makePromiseContract(); try { + SLogger::info("DBG notify: before scope.add"); // The send callback may reach into the transport (it captures // the communicator owner), so it runs as an mScope task — the // scope drain in the destructor replaces the former private @@ -318,11 +320,13 @@ public: mOutgoingMessageCallback]() mutable -> folly::coro::Task { try { + SLogger::info("DBG notify-task: sending"); outgoingMessageCallback( requestMessage, requestMessage.requestid(), callContext); promise.setValue(); + SLogger::info("DBG notify-task: sent"); } catch ( const std::exception& clientSideException) { SLogger::debug( @@ -334,6 +338,7 @@ public: } co_return; }))); + SLogger::info("DBG notify: awaiting contract"); co_return co_await folly::coro::timeout( folly::coro::detachOnCancel(std::move(promiseContract.second)), timeoutValue); diff --git a/packages/streamr-utils/modules/GuardedAsyncScope.cppm b/packages/streamr-utils/modules/GuardedAsyncScope.cppm index d4518c49..9b827b64 100644 --- a/packages/streamr-utils/modules/GuardedAsyncScope.cppm +++ b/packages/streamr-utils/modules/GuardedAsyncScope.cppm @@ -18,6 +18,7 @@ module; export module streamr.utils.GuardedAsyncScope; +import streamr.logger.SLogger; import streamr.utils.CoroutineHelper; export namespace streamr::utils { @@ -39,7 +40,9 @@ public: if (this->closed) { return; } + streamr::logger::SLogger::info("DBG GuardedAsyncScope: adding"); this->scope.add(std::forward(awaitable)); + streamr::logger::SLogger::info("DBG GuardedAsyncScope: added"); } // Blocks until every added task has finished; further adds are From 2ba8ddbff9d488e6030ea9bf9055583169c51d10 Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Sat, 11 Jul 2026 01:00:02 +0300 Subject: [PATCH 6/6] Revert "DEBUG: instrument the reverse-connection notify path (temporary)" This reverts commit 2f5bcbe431139936669039b2f836b8db08ae3d59. --- .../connection/websocket/WebsocketServerConnector.cppm | 9 +++------ .../modules/RpcCommunicatorClientApi.cppm | 7 +------ packages/streamr-utils/modules/GuardedAsyncScope.cppm | 3 --- 3 files changed, 4 insertions(+), 15 deletions(-) diff --git a/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm b/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm index f6b220c2..08a41c03 100644 --- a/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm +++ b/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm @@ -411,7 +411,6 @@ private: this->ongoingConnectRequests.erase(nodeId); }); this->ongoingConnectRequests.emplace(nodeId, pendingConnection); - SLogger::info("DBG rcfp: before scope.add"); this->requestConnectionScope.add( streamr::utils::co_withExecutor( &this->requestConnectionExecutor, @@ -419,23 +418,21 @@ private: [this, localPeerDescriptor, targetPeerDescriptor]() -> folly::coro::Task { try { - SLogger::info("DBG rcfp-task: entered"); WebsocketClientConnectorRpcClient client( this->options.rpcCommunicator); - SLogger::info("DBG rcfp-task: client built"); WebsocketClientConnectorRpcRemote remoteConnector( PeerDescriptor(localPeerDescriptor), PeerDescriptor(targetPeerDescriptor), std::move(client)); - SLogger::info("DBG rcfp-task: remote built"); // Cancellable by destroy(): abort fires before the // executor join drains this task. co_await streamr::utils::co_withCancellation( this->abortController.getSignal() .getCancellationToken(), remoteConnector.requestConnection()); - SLogger::info( - "DBG rcfp-task: notify co_await done"); + SLogger::trace( + "Sent WebsocketConnectionRequest notification" + " to peer"); } catch (const std::exception& err) { SLogger::debug( "Failed to send WebsocketConnectionRequest" diff --git a/packages/streamr-proto-rpc/modules/RpcCommunicatorClientApi.cppm b/packages/streamr-proto-rpc/modules/RpcCommunicatorClientApi.cppm index 0068cf11..c7e993fa 100644 --- a/packages/streamr-proto-rpc/modules/RpcCommunicatorClientApi.cppm +++ b/packages/streamr-proto-rpc/modules/RpcCommunicatorClientApi.cppm @@ -288,7 +288,7 @@ public: const RequestType& notificationParam, const CallContextType& callContext, std::optional timeout = std::nullopt) { - SLogger::info("DBG notify: entered"); + SLogger::trace("notify() notificationName:", notificationName); std::chrono::milliseconds timeoutValue = mRpcRequestTimeout; if (timeout.has_value()) { @@ -300,11 +300,9 @@ public: notificationName); auto requestMessage = this->createRequestRpcMessage( notificationName, notificationParam, true); - SLogger::info("DBG notify: message created"); auto&& promiseContract = folly::coro::makePromiseContract(); try { - SLogger::info("DBG notify: before scope.add"); // The send callback may reach into the transport (it captures // the communicator owner), so it runs as an mScope task — the // scope drain in the destructor replaces the former private @@ -320,13 +318,11 @@ public: mOutgoingMessageCallback]() mutable -> folly::coro::Task { try { - SLogger::info("DBG notify-task: sending"); outgoingMessageCallback( requestMessage, requestMessage.requestid(), callContext); promise.setValue(); - SLogger::info("DBG notify-task: sent"); } catch ( const std::exception& clientSideException) { SLogger::debug( @@ -338,7 +334,6 @@ public: } co_return; }))); - SLogger::info("DBG notify: awaiting contract"); co_return co_await folly::coro::timeout( folly::coro::detachOnCancel(std::move(promiseContract.second)), timeoutValue); diff --git a/packages/streamr-utils/modules/GuardedAsyncScope.cppm b/packages/streamr-utils/modules/GuardedAsyncScope.cppm index 9b827b64..d4518c49 100644 --- a/packages/streamr-utils/modules/GuardedAsyncScope.cppm +++ b/packages/streamr-utils/modules/GuardedAsyncScope.cppm @@ -18,7 +18,6 @@ module; export module streamr.utils.GuardedAsyncScope; -import streamr.logger.SLogger; import streamr.utils.CoroutineHelper; export namespace streamr::utils { @@ -40,9 +39,7 @@ public: if (this->closed) { return; } - streamr::logger::SLogger::info("DBG GuardedAsyncScope: adding"); this->scope.add(std::forward(awaitable)); - streamr::logger::SLogger::info("DBG GuardedAsyncScope: added"); } // Blocks until every added task has finished; further adds are