diff --git a/.clangd b/.clangd index a3c2d24c..cf867c8b 100644 --- a/.clangd +++ b/.clangd @@ -5,3 +5,9 @@ InlayHints: ParameterNames: No DeducedTypes: Yes Designators: No +# No background index: lint.sh runs several short-lived clangd processes +# back-to-back (clangd-tidy chunks), and a clangd starting while the previous +# one's index shards are mid-write crashes at startup (clangd-tidy aborts with +# "Invalid header end"). The lint runs don't benefit from the index anyway. +Index: + Background: Skip diff --git a/packages/streamr-dht/CMakeLists.txt b/packages/streamr-dht/CMakeLists.txt index a92e015f..092eb266 100644 --- a/packages/streamr-dht/CMakeLists.txt +++ b/packages/streamr-dht/CMakeLists.txt @@ -254,6 +254,12 @@ if(NOT IOS AND STREAMR_MODULES_SUPPORTED) test/integration/ConnectionManagerIntegrationTest.cpp test/integration/DhtNodeRpcRemoteTest.cpp test/integration/RouterRpcRemoteTest.cpp + test/integration/MultipleEntryPointJoiningTest.cpp + test/integration/DhtNodeTest.cpp + test/integration/MockLayer1Layer0Test.cpp + test/integration/Layer1ScaleTest.cpp + test/integration/DhtNodeExternalApiTest.cpp + test/integration/KademliaCorrectnessTest.cpp ) target_include_directories(streamr-dht-test-integration diff --git a/packages/streamr-dht/lint.sh b/packages/streamr-dht/lint.sh index 0b968556..26914f36 100755 --- a/packages/streamr-dht/lint.sh +++ b/packages/streamr-dht/lint.sh @@ -50,8 +50,48 @@ echo "Running clangd-tidy on $FILES" # false positive through the StoreRpcRemote / StoreManager coroutine # cluster; the compiler builds and runs both, clang-format still checks # them. (LocalDataStoreTest.cpp does NOT trip it.) -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' | tr '\n' ' ') -clangd-tidy -p "$COMPILE_DB" $TIDY_FILES +# MultipleEntryPointJoiningTest.cpp (phase A8) trips the same false +# positive through the DhtNode import set (a spurious "no matching +# constructor" for the branded ServiceID inside the included +# DhtNodeTestUtils.hpp); the compiler builds and runs it, and the OTHER +# A8 tests including the same header pass clangd — the variance is the +# hallmark of the std-type unification issue. clang-format still checks it. +# DhtNodeTestUtils.hpp (phase A8) is a plain header whose names come from +# module imports done by the INCLUDING test (a module wrapping DhtNode's +# interface exhausts clang's source locations, so it cannot be a module — +# see the header comment); standalone clangd has no imports to resolve it +# against, so every name errors (and processing it standalone can crash +# clangd, aborting the whole batch). The including tests are clangd-tidied +# and the compiler builds it on every platform; clang-format still checks +# it. +# The full-node integration tests (phase A8) each import the DhtNode +# aggregate, whose BMI merge consumes most of clang's per-TU +# source-location space (see module-sloc-dedup-plan.md); clangd does not +# fully release that space between files, so batching several of them into +# ONE clangd process exhausts it and kills clangd mid-session (clangd-tidy +# aborts with "Invalid header end"). They lint clean individually, so each +# gets its own clangd process below. (DhtNodeTest.cpp is excluded outright +# instead: it trips the std-type unification false positive — gtest's own +# CodeLocation/MakeAndRegisterTestInfo reported ambiguous — like the other +# excluded files; the compiler builds and runs it on every platform and +# clang-format still checks 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' ' ') +# 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 +# files degrade to non-module parses — spurious "unknown type name 'import'" +# — and clangd finally dies, aborting clangd-tidy with "Invalid header end"). +# Verified: every 13-file chunk passes clean where the one-shot batch dies. +# The `< /dev/null` matters: a clangd-tidy spawned by xargs inherits the +# exhausted file-list pipe as stdin, which kills clangd at startup (verified +# both ways). +echo "$TIDY_FILES" | tr ' ' '\n' | grep -v '^$' | \ + xargs -n 13 sh -c 'clangd-tidy -p "$0" "$@" < /dev/null' "$COMPILE_DB" +for heavy_file in $HEAVY_TIDY_FILES; do + echo "Running clangd-tidy (own process) on $heavy_file" + clangd-tidy -p "$COMPILE_DB" "$heavy_file" < /dev/null +done echo "Running clang-format --dry-run on $FILES" ../../run-clang-format.py $FILES diff --git a/packages/streamr-dht/modules/connection/ConnectionManager.cppm b/packages/streamr-dht/modules/connection/ConnectionManager.cppm index dd232e27..77cc4c9f 100644 --- a/packages/streamr-dht/modules/connection/ConnectionManager.cppm +++ b/packages/streamr-dht/modules/connection/ConnectionManager.cppm @@ -573,13 +573,21 @@ public: endpointsSnapshot = this->endpoints | std::views::values | std::ranges::to(); } - return endpointsSnapshot | std::views::filter([](const auto& endpoint) { - return endpoint->isConnected(); - }) | - std::views::transform([](const auto& endpoint) { - return endpoint->getPeerDescriptor(); - }) | - std::ranges::to(); + // Materialized with a single-pass loop, NOT a views::filter | + // ranges::to pipeline: for a forward range libc++'s to walks + // the range twice (ranges::distance to size the allocation, then the + // copy), and isConnected() is live state — an endpoint connecting + // between the two passes made the copy pass yield MORE elements than + // were counted, overflowing the allocation (heap-buffer-overflow + // caught by ASan in the 48-node layer1-scale test). + std::vector connections; + connections.reserve(endpointsSnapshot.size()); + for (const auto& endpoint : endpointsSnapshot) { + if (endpoint->isConnected()) { + connections.push_back(endpoint->getPeerDescriptor()); + } + } + return connections; } private: diff --git a/packages/streamr-dht/modules/dht/DhtNode.cppm b/packages/streamr-dht/modules/dht/DhtNode.cppm new file mode 100644 index 00000000..ccedba4b --- /dev/null +++ b/packages/streamr-dht/modules/dht/DhtNode.cppm @@ -0,0 +1,738 @@ +// Module streamr.dht.DhtNode +// Ported from packages/dht/src/dht/DhtNode.ts (v103.8.0-rc.3). The full DHT +// node: it composes PeerManager, PeerDiscovery, Router, +// RecursiveOperationManager and StoreManager over one RoutingRpcCommunicator, +// and implements the Transport interface so a layer-1 DhtNode can use a +// layer-0 DhtNode as its transport. +// +// Scope of this port (phase A8): only the "transport given" path — every test +// constructs a DhtNode over a SimulatorTransport (or, for layer 1, over a +// layer-0 DhtNode). The ConnectionManager/DefaultConnectorFacade path (real +// sockets, connectivity checking, createPeerDescriptor from a connectivity +// response, CDN/GeoIP region) is real-socket territory deferred to milestone B; +// start() throws if no transport is given. The external DhtNode contact events +// (nearbyContactAdded etc.) have no subscriber below milestone C, so Store +// replication and the k-bucket-empty rejoin are wired straight to PeerManager's +// own events instead of being re-emitted; periodic neighbour pinging (an opt-in +// no A8 test enables) is likewise deferred. +module; + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include // IWYU pragma: keep + +export module streamr.dht.DhtNode; + +import streamr.dht.protos; + +import streamr.eventemitter.EventEmitter; +import streamr.utils.AbortController; +import streamr.utils.CoroutineHelper; +import streamr.utils.ExecutorHelper; +import streamr.utils.waitForCondition; +import streamr.logger.SLogger; +import streamr.dht.ConnectionLocker; +import streamr.dht.ConnectionsView; +import streamr.dht.consts; +import streamr.dht.DhtCallContext; +import streamr.dht.DhtNodeRpcLocal; +import streamr.dht.DhtNodeRpcRemote; +import streamr.dht.DhtRpcClient; +import streamr.dht.ExternalApiRpcLocal; +import streamr.dht.ExternalApiRpcRemote; +import streamr.dht.Identifiers; +import streamr.dht.LocalDataStore; +import streamr.dht.PeerDiscovery; +import streamr.dht.PeerManager; +import streamr.dht.RecursiveOperationManager; +import streamr.dht.RecursiveOperationSession; +import streamr.dht.ringIdentifiers; +import streamr.dht.Router; +import streamr.dht.RoutingSession; +import streamr.dht.RoutingRpcCommunicator; +import streamr.dht.StoreManager; +import streamr.dht.StoreRpcRemote; +import streamr.dht.Transport; +import streamr.protorpc.RpcCommunicator; + +// Hoisted from the former header (file scope, NOT exported). +using streamr::eventemitter::HandlerToken; +using streamr::logger::SLogger; +using streamr::utils::AbortController; +using streamr::utils::waitForCondition; + +export namespace streamr::dht { + +using ::dht::ClosestPeersRequest; +using ::dht::ClosestPeersResponse; +using ::dht::ClosestRingPeersRequest; +using ::dht::ClosestRingPeersResponse; +using ::dht::DataEntry; +using ::dht::ExternalFetchDataRequest; +using ::dht::ExternalFetchDataResponse; +using ::dht::ExternalStoreDataRequest; +using ::dht::ExternalStoreDataResponse; +using ::dht::LeaveNotice; +using ::dht::Message; +using ::dht::PeerDescriptor; +using ::dht::PingRequest; +using ::dht::PingResponse; +using ::dht::RecursiveOperation; +using ::dht::RouteMessageWrapper; +using streamr::dht::connection::ConnectionLocker; +using streamr::dht::connection::ConnectionsView; +using streamr::dht::connection::LockID; +using streamr::dht::contact::RingIdRaw; +using streamr::dht::discovery::PeerDiscovery; +using streamr::dht::discovery::PeerDiscoveryOptions; +using streamr::dht::recursiveoperation::RecursiveOperationManager; +using streamr::dht::recursiveoperation::RecursiveOperationManagerOptions; +using streamr::dht::recursiveoperation::RecursiveOperationResult; +using streamr::dht::routing::Router; +using streamr::dht::routing::RouterOptions; +using streamr::dht::routing::RoutingMode; +using streamr::dht::rpcprotocol::DhtCallContext; +using streamr::dht::store::LocalDataStore; +using streamr::dht::store::StoreManager; +using streamr::dht::store::StoreManagerOptions; +using streamr::dht::store::StoreRpcRemote; +using streamr::dht::transport::RoutingRpcCommunicator; +using streamr::dht::transport::SendOptions; +using streamr::dht::transport::Transport; +using streamr::dht::transport::transportevents::Connected; +using streamr::dht::transport::transportevents::Disconnected; +namespace transportevents = streamr::dht::transport::transportevents; +using streamr::dht::store::StoreRpcClient; +using streamr::protorpc::RpcCommunicatorOptions; +// DhtNodeRpcClient and ExternalApiRpcClient are exported into streamr::dht by +// the imported DhtNodeRpcRemote / ExternalApiRpcRemote modules. + +inline constexpr size_t numberOfNodesPerKBucketDefault = 8; + +struct DhtNodeOptions { + ServiceID serviceId = CONTROL_LAYER_NODE_SERVICE_ID; + size_t joinParallelism = 3; + size_t maxContactCount = 200; + size_t numberOfNodesPerKBucket = numberOfNodesPerKBucketDefault; + size_t joinNoProgressLimit = 5; + std::chrono::milliseconds dhtJoinTimeout{60000}; + size_t peerDiscoveryQueryBatchSize = 5; + uint32_t storeHighestTtl = 60000; + uint32_t storeMaxTtl = 60000; + std::chrono::milliseconds networkConnectivityTimeout{10000}; + size_t storageRedundancyFactor = 5; + std::optional neighborPingLimit; + std::optional rpcRequestTimeout; + + // Given transport (required in this phase). The three roles are the same + // object for a SimulatorTransport / ConnectionManager. + Transport* transport = nullptr; + ConnectionsView* connectionsView = nullptr; + ConnectionLocker* connectionLocker = nullptr; // may be null + std::optional peerDescriptor; + std::vector entryPoints; +}; + +class DhtNode : public Transport { +private: + static constexpr std::chrono::milliseconds externalApiTimeout{10000}; + static constexpr std::chrono::milliseconds networkConnectivityPollInterval{ + 100}; + + DhtNodeOptions options; + LocalDataStore localDataStore; + std::unique_ptr rpcCommunicator; + Transport* transportPtr = nullptr; + ConnectionsView* connectionsView = nullptr; + ConnectionLocker* connectionLocker = nullptr; + std::shared_ptr + connectionLockerShared; // non-owning alias + std::optional localPeerDescriptor; + std::shared_ptr router; + std::shared_ptr storeManager; + std::shared_ptr recursiveOperationManager; + std::shared_ptr peerDiscovery; + std::shared_ptr peerManager; + std::unique_ptr dhtNodeRpcLocal; + std::shared_ptr externalApiRpcLocal; + bool started = false; + AbortController abortController; + // Detaches the k-bucket-empty rejoin off the delivery thread that raised + // the event. + folly::CPUThreadPoolExecutor recoveryExecutor{1}; + HandlerToken messageToken; + HandlerToken connectedToken; + HandlerToken disconnectedToken; + + [[nodiscard]] std::shared_ptr createDhtNodeRpcRemote( + const PeerDescriptor& peerDescriptor) { + return std::make_shared( + *this->localPeerDescriptor, + peerDescriptor, + this->options.serviceId, + DhtNodeRpcClient(*this->rpcCommunicator), + this->options.rpcRequestTimeout); + } + + // PeerManager::getNeighbors returns the k-bucket remotes; the callers here + // (and the public getNeighbors) want their descriptors (TS maps them). + [[nodiscard]] std::vector getNeighborDescriptors() { + std::vector result; + for (const auto& neighbor : this->peerManager->getNeighbors()) { + result.push_back(neighbor->getPeerDescriptor()); + } + return result; + } + + void handleMessageFromTransport(const Message& message) { + if (message.serviceid() == this->options.serviceId) { + this->rpcCommunicator->handleMessageFromPeer(message); + } + } + + void handleMessageFromRouter(const Message& message) { + if (message.serviceid() == this->options.serviceId) { + this->rpcCommunicator->handleMessageFromPeer(message); + } else { + this->emit(message); + } + } + + [[nodiscard]] std::vector getConnectedEntryPoints() { + std::vector result; + for (const auto& entryPoint : this->options.entryPoints) { + if (this->connectionsView->hasConnection( + Identifiers::getNodeIdFromPeerDescriptor(entryPoint))) { + result.push_back(entryPoint); + } + } + return result; + } + + template + folly::coro::Task executeRecursiveOperation( + std::function()> executeDirectly, + std::function(const PeerDescriptor&)> + executeViaPeer) { + const auto connectedEntryPoints = this->getConnectedEntryPoints(); + if (this->peerDiscovery->isJoinOngoing() && + !connectedEntryPoints.empty()) { + // TS uses lodash sample (a random entry point); the first is + // sufficient and deterministic for the tests. + co_return co_await executeViaPeer(connectedEntryPoints.front()); + } + co_return co_await executeDirectly(); + } + + void initPeerManager() { + this->peerManager = PeerManager::newInstance( + PeerManagerOptions{ + .numberOfNodesPerKBucket = + this->options.numberOfNodesPerKBucket, + .maxContactCount = this->options.maxContactCount, + .localNodeId = this->getNodeId(), + .localPeerDescriptor = *this->localPeerDescriptor, + .connectionLocker = this->connectionLockerShared, + .neighborPingLimit = this->options.neighborPingLimit, + .lockId = LockID{this->options.serviceId}, + .createDhtNodeRpcRemote = + [this](const PeerDescriptor& peerDescriptor) { + return this->createDhtNodeRpcRemote(peerDescriptor); + }, + .hasConnection = + [this](const DhtAddress& nodeId) { + return this->connectionsView->hasConnection(nodeId); + }}); + // StoreManager tracks the neighbourhood via nearbyContactAdded (TS + // re-emits it as a DhtNode event first; wired straight through here). + this->peerManager->on( + [this](const PeerDescriptor& peerDescriptor) { + this->storeManager->onContactAdded(peerDescriptor); + }); + this->peerManager->on([this]() { + if (!this->peerDiscovery->isJoinOngoing() && + !this->options.entryPoints.empty()) { + // Pin the discovery (shared_ptr) and cancel with the node's + // abort signal so the detached rejoin is safe across stop(). + auto discovery = this->peerDiscovery; + const auto token = + this->abortController.getSignal().getCancellationToken(); + for (const auto& entryPoint : this->options.entryPoints) { + streamr::utils::co_withExecutor( + &this->recoveryExecutor, + streamr::utils::co_withCancellation( + token, + folly::coro::co_invoke( + [discovery, + entryPoint]() -> folly::coro::Task { + co_await discovery->rejoinDht(entryPoint); + }))) + .start(); + } + } + }); + this->connectedToken = this->transportPtr->on( + [this](const PeerDescriptor& peerDescriptor) { + this->router->onNodeConnected(peerDescriptor); + this->emit(peerDescriptor); + }); + this->disconnectedToken = this->transportPtr->on( + [this](const PeerDescriptor& peerDescriptor, bool gracefulLeave) { + if (this->connectionLocker != nullptr) { + const auto nodeId = + Identifiers::getNodeIdFromPeerDescriptor( + peerDescriptor); + if (gracefulLeave) { + this->peerManager->removeContact(nodeId); + } else { + this->peerManager->removeNeighbor(nodeId); + } + } + this->router->onNodeDisconnected(peerDescriptor); + this->emit(peerDescriptor, gracefulLeave); + }); + } + + void bindRpcLocalMethods() { + this->dhtNodeRpcLocal = + std::make_unique(DhtNodeRpcLocalOptions{ + .peerDiscoveryQueryBatchSize = + this->options.peerDiscoveryQueryBatchSize, + .getNeighbors = + [this]() { return this->getNeighborDescriptors(); }, + .getClosestRingContactsTo = + [this](const RingIdRaw& ringIdRaw, size_t limit) { + return this->getClosestRingContactsTo(ringIdRaw, limit); + }, + .addContact = + [this](const PeerDescriptor& contact) { + this->peerManager->addContact(contact); + }, + .removeContact = + [this](const DhtAddress& nodeId) { + this->removeContact(nodeId); + }}); + this->rpcCommunicator + ->registerRpcMethod( + "getClosestPeers", + [this]( + const ClosestPeersRequest& req, + const DhtCallContext& context) { + return this->dhtNodeRpcLocal->getClosestPeers(req, context); + }); + this->rpcCommunicator->registerRpcMethod< + ClosestRingPeersRequest, + ClosestRingPeersResponse>( + "getClosestRingPeers", + [this]( + const ClosestRingPeersRequest& req, + const DhtCallContext& context) { + return this->dhtNodeRpcLocal->getClosestRingPeers(req, context); + }); + this->rpcCommunicator->registerRpcMethod( + "ping", + [this](const PingRequest& req, const DhtCallContext& context) { + return this->dhtNodeRpcLocal->ping(req, context); + }); + this->rpcCommunicator->registerRpcNotification( + "leaveNotice", + [this](const LeaveNotice& req, const DhtCallContext& context) { + this->dhtNodeRpcLocal->leaveNotice(req, context); + }); + this->externalApiRpcLocal = + std::make_shared(ExternalApiRpcLocalOptions{ + .executeRecursiveOperation = + [this]( + const DhtAddress& key, + RecursiveOperation operation, + const DhtAddress& excludedPeer) { + return this->recursiveOperationManager->execute( + key, operation, excludedPeer); + }, + .storeDataToDht = + [this]( + const DhtAddress& key, + const ::google::protobuf::Any& data, + const DhtAddress& creator) { + return this->storeDataToDht(key, data, creator); + }}); + auto externalApi = this->externalApiRpcLocal; + this->rpcCommunicator->registerRpcMethodAsync< + ExternalFetchDataRequest, + ExternalFetchDataResponse>( + "externalFetchData", + [externalApi]( + const ExternalFetchDataRequest& req, + const DhtCallContext& context) + -> folly::coro::Task { + co_return co_await externalApi->externalFetchData(req, context); + }); + this->rpcCommunicator->registerRpcMethodAsync< + ExternalStoreDataRequest, + ExternalStoreDataResponse>( + "externalStoreData", + [externalApi]( + const ExternalStoreDataRequest& req, + const DhtCallContext& context) + -> folly::coro::Task { + co_return co_await externalApi->externalStoreData(req, context); + }); + } + +public: + explicit DhtNode(DhtNodeOptions options) + : options(std::move(options)), + localDataStore(this->options.storeMaxTtl) {} + + ~DhtNode() override = default; + DhtNode(const DhtNode&) = delete; + DhtNode& operator=(const DhtNode&) = delete; + DhtNode(DhtNode&&) = delete; + DhtNode& operator=(DhtNode&&) = delete; + + folly::coro::Task start() { + if (this->started || this->abortController.getSignal().aborted) { + co_return; + } + this->started = true; + if (this->options.transport == nullptr) { + throw std::runtime_error( + "DhtNode requires a transport in this build (the " + "ConnectionManager path arrives in milestone B)"); + } + this->transportPtr = this->options.transport; + this->connectionsView = this->options.connectionsView; + this->connectionLocker = this->options.connectionLocker; + if (this->connectionLocker != nullptr) { + this->connectionLockerShared = std::shared_ptr( + this->connectionLocker, [](ConnectionLocker*) {}); + } + this->localPeerDescriptor = + this->transportPtr->getLocalPeerDescriptor(); + + std::optional communicatorOptions; + if (this->options.rpcRequestTimeout.has_value()) { + communicatorOptions = RpcCommunicatorOptions{ + .rpcRequestTimeout = *this->options.rpcRequestTimeout}; + } + this->rpcCommunicator = std::make_unique( + ServiceID{this->options.serviceId}, + [this](Message msg, SendOptions sendOptions) { + this->transportPtr->send(msg, sendOptions); + }, + std::move(communicatorOptions)); + + this->messageToken = this->transportPtr->on( + [this](const Message& message) { + this->handleMessageFromTransport(message); + }); + + this->initPeerManager(); + + this->peerDiscovery = PeerDiscovery::newInstance( + PeerDiscoveryOptions{ + .localPeerDescriptor = *this->localPeerDescriptor, + .joinNoProgressLimit = this->options.joinNoProgressLimit, + .serviceId = this->options.serviceId, + .parallelism = this->options.joinParallelism, + .joinTimeout = this->options.dhtJoinTimeout, + .connectionLocker = this->connectionLockerShared, + .peerManager = *this->peerManager, + .abortSignal = this->abortController.getSignal(), + .createDhtNodeRpcRemote = + [this](const PeerDescriptor& peerDescriptor) { + return this->createDhtNodeRpcRemote(peerDescriptor); + }}); + this->router = Router::newInstance( + RouterOptions{ + .rpcCommunicator = *this->rpcCommunicator, + .localPeerDescriptor = *this->localPeerDescriptor, + .handleMessage = + [this](const Message& message) { + this->handleMessageFromRouter(message); + }, + .getConnections = + [this]() { + return this->connectionsView->getConnections(); + }}); + auto routerPtr = this->router; + this->recursiveOperationManager = + RecursiveOperationManager::newInstance( + RecursiveOperationManagerOptions{ + .rpcCommunicator = *this->rpcCommunicator, + .sessionTransport = *this, + .connectionsView = *this->connectionsView, + .localPeerDescriptor = *this->localPeerDescriptor, + .serviceId = this->options.serviceId, + .localDataStore = this->localDataStore, + .addContact = + [this](const PeerDescriptor& contact) { + this->peerManager->addContact(contact); + }, + .createDhtNodeRpcRemote = + [this](const PeerDescriptor& peerDescriptor) { + return this->createDhtNodeRpcRemote(peerDescriptor); + }, + .doRouteMessage = + [routerPtr]( + const RouteMessageWrapper& message, + RoutingMode mode, + const std::optional& excludedPeer) { + return routerPtr->doRouteMessage( + message, mode, excludedPeer); + }, + .isMostLikelyDuplicate = + [routerPtr](const std::string& requestId) { + return routerPtr->isMostLikelyDuplicate(requestId); + }, + .addToDuplicateDetector = + [routerPtr](const std::string& requestId) { + routerPtr->addToDuplicateDetector(requestId); + }}); + this->storeManager = StoreManager::newInstance( + StoreManagerOptions{ + .rpcCommunicator = *this->rpcCommunicator, + .localPeerDescriptor = *this->localPeerDescriptor, + .localDataStore = this->localDataStore, + .serviceId = this->options.serviceId, + .highestTtl = this->options.storeHighestTtl, + .redundancyFactor = this->options.storageRedundancyFactor, + .getNeighbors = + [this]() { return this->getNeighborDescriptors(); }, + .createRpcRemote = + [this](const PeerDescriptor& contact) { + return std::make_shared( + *this->localPeerDescriptor, + contact, + StoreRpcClient(*this->rpcCommunicator), + this->options.rpcRequestTimeout); + }, + .executeRecursiveOperation = + [this]( + const DhtAddress& key, RecursiveOperation operation) { + return this->recursiveOperationManager->execute( + key, operation); + }}); + this->bindRpcLocalMethods(); + co_return; + } + + // --- Transport interface --------------------------------------------- + + void send( + const Message& message, const SendOptions& /*sendOptions*/) override { + if (!this->started || this->abortController.getSignal().aborted) { + return; + } + const auto reachableThrough = this->peerDiscovery->isJoinOngoing() + ? this->getConnectedEntryPoints() + : std::vector{}; + this->router->send(message, reachableThrough); + } + + [[nodiscard]] PeerDescriptor getLocalPeerDescriptor() const override { + return *this->localPeerDescriptor; + } + + void stop() override { + if (this->abortController.getSignal().aborted || !this->started) { + return; + } + SLogger::trace("DhtNode::stop()"); + this->abortController.abort(); + this->transportPtr->off(this->messageToken); + this->transportPtr->off(this->connectedToken); + this->transportPtr->off(this->disconnectedToken); + streamr::utils::blockingWait(this->storeManager->destroy()); + this->localDataStore.clear(); + if (this->peerManager) { + this->peerManager->stop(); + } + this->router->stop(); + this->recursiveOperationManager->stop(); + // The transport was given, not created here, so it is not stopped. + this->connectionLocker = nullptr; + } + + // --- Public API ------------------------------------------------------ + + folly::coro::Task joinDht( + std::vector entryPointDescriptors, + bool doAdditionalDistantPeerDiscovery = true, + bool retry = true) { + if (!this->started) { + throw std::runtime_error( + "Cannot join DHT before calling start() on DhtNode"); + } + co_await this->peerDiscovery->joinDht( + std::move(entryPointDescriptors), + doAdditionalDistantPeerDiscovery, + retry); + } + + folly::coro::Task joinRing() { + if (!this->started) { + throw std::runtime_error( + "Cannot join ring before calling start() on DhtNode"); + } + co_await this->peerDiscovery->joinRing(); + } + + folly::coro::Task> storeDataToDht( + const DhtAddress& key, + const ::google::protobuf::Any& data, + std::optional creator = std::nullopt) { + const DhtAddress effectiveCreator = creator.value_or(this->getNodeId()); + co_return co_await this + ->executeRecursiveOperation>( + [this, key, data, effectiveCreator]() + -> folly::coro::Task> { + co_return co_await this->storeManager->storeDataToDht( + key, data, effectiveCreator); + }, + [this, key, data](const PeerDescriptor& connectedEntryPoint) + -> folly::coro::Task> { + co_return co_await this->storeDataToDhtViaPeer( + key, data, connectedEntryPoint); + }); + } + + folly::coro::Task> storeDataToDhtViaPeer( + const DhtAddress& key, + const ::google::protobuf::Any& data, + PeerDescriptor peer) { + ExternalApiRpcRemote rpcRemote( + *this->localPeerDescriptor, + std::move(peer), + ExternalApiRpcClient(*this->rpcCommunicator)); + co_return co_await rpcRemote.storeData(key, data); + } + + folly::coro::Task> fetchDataFromDht( + const DhtAddress& key) { + co_return co_await this + ->executeRecursiveOperation>( + [this, key]() -> folly::coro::Task> { + const auto result = + co_await this->recursiveOperationManager->execute( + key, RecursiveOperation::FETCH_DATA); + co_return result.dataEntries; + }, + [this, key](const PeerDescriptor& connectedEntryPoint) + -> folly::coro::Task> { + co_return co_await this->fetchDataFromDhtViaPeer( + key, connectedEntryPoint); + }); + } + + folly::coro::Task> fetchDataFromDhtViaPeer( + const DhtAddress& key, PeerDescriptor peer) { + ExternalApiRpcRemote rpcRemote( + *this->localPeerDescriptor, + std::move(peer), + ExternalApiRpcClient(*this->rpcCommunicator)); + co_return co_await rpcRemote.externalFetchData(key); + } + + folly::coro::Task> findClosestNodesFromDht( + const DhtAddress& key) { + co_return co_await this + ->executeRecursiveOperation>( + [this, + key]() -> folly::coro::Task> { + const auto result = + co_await this->recursiveOperationManager->execute( + key, RecursiveOperation::FIND_CLOSEST_NODES); + co_return result.closestNodes; + }, + [this, key](const PeerDescriptor& connectedEntryPoint) + -> folly::coro::Task> { + ExternalApiRpcRemote rpcRemote( + *this->localPeerDescriptor, + connectedEntryPoint, + ExternalApiRpcClient(*this->rpcCommunicator)); + co_return co_await rpcRemote.externalFindClosestNode(key); + }); + } + + folly::coro::Task waitForNetworkConnectivity() { + co_await waitForCondition( + [this]() { + return this->connectionsView->getConnectionCount() > 0; + }, + this->options.networkConnectivityTimeout, + networkConnectivityPollInterval, + &this->abortController.getSignal()); + } + + [[nodiscard]] DhtAddress getNodeId() { + return Identifiers::getNodeIdFromPeerDescriptor( + *this->localPeerDescriptor); + } + + [[nodiscard]] size_t getNeighborCount() { + return this->peerManager->getNeighborCount(); + } + + [[nodiscard]] std::vector getNeighbors() { + if (!this->started) { + return {}; + } + return this->getNeighborDescriptors(); + } + + [[nodiscard]] std::vector getClosestContacts( + std::optional limit = std::nullopt) { + return this->peerManager->getClosestContacts(limit); + } + + [[nodiscard]] std::vector getRandomContacts( + std::optional limit = std::nullopt) { + return this->peerManager->getRandomContacts(limit); + } + + [[nodiscard]] ClosestRingPeerDescriptors getRingContacts() { + return this->peerManager->getRingContacts(); + } + + [[nodiscard]] ClosestRingPeerDescriptors getClosestRingContactsTo( + const RingIdRaw& ringIdRaw, + std::optional limit = std::nullopt) { + const auto closest = + this->peerManager->getClosestRingContactsTo(ringIdRaw, limit); + ClosestRingPeerDescriptors result; + for (const auto& contact : closest.left) { + result.left.push_back(contact->getPeerDescriptor()); + } + for (const auto& contact : closest.right) { + result.right.push_back(contact->getPeerDescriptor()); + } + return result; + } + + void removeContact(const DhtAddress& nodeId) { + if (!this->started) { + return; + } + this->peerManager->removeContact(nodeId); + } + + [[nodiscard]] bool hasJoined() { + return this->peerDiscovery->isJoinCalled(); + } + + [[nodiscard]] ConnectionsView* getConnectionsView() { + return this->connectionsView; + } +}; + +} // namespace streamr::dht diff --git a/packages/streamr-dht/modules/dht/ExternalApiRpcLocal.cppm b/packages/streamr-dht/modules/dht/ExternalApiRpcLocal.cppm new file mode 100644 index 00000000..9283ec9d --- /dev/null +++ b/packages/streamr-dht/modules/dht/ExternalApiRpcLocal.cppm @@ -0,0 +1,106 @@ +// Module streamr.dht.ExternalApiRpcLocal +// Ported from packages/dht/src/dht/ExternalApiRpcLocal.ts (v103.8.0-rc.3). +// The server side of the ExternalApi service: it lets a peer run a recursive +// fetch/store/find-closest on this node's behalf (used while that peer is +// still joining and can only reach the DHT through a connected entry point). +// +// The handlers are coroutines, not implementations of the generated +// (synchronous) ExternalApiRpc interface: each runs a recursive DHT operation +// that sends messages and awaits their responses, so it must SUSPEND rather +// than block — DhtNode registers them with registerRpcMethodAsync. +module; + +#include +#include + +#include // IWYU pragma: keep + +export module streamr.dht.ExternalApiRpcLocal; + +import streamr.dht.protos; + +import streamr.utils.CoroutineHelper; +import streamr.dht.DhtCallContext; +import streamr.dht.Identifiers; +import streamr.dht.RecursiveOperationSession; + +export namespace streamr::dht { + +using ::dht::ExternalFetchDataRequest; +using ::dht::ExternalFetchDataResponse; +using ::dht::ExternalFindClosestNodesRequest; +using ::dht::ExternalFindClosestNodesResponse; +using ::dht::ExternalStoreDataRequest; +using ::dht::ExternalStoreDataResponse; +using ::dht::PeerDescriptor; +using ::dht::RecursiveOperation; +using streamr::dht::Identifiers; +using streamr::dht::recursiveoperation::RecursiveOperationResult; +using streamr::dht::rpcprotocol::DhtCallContext; + +struct ExternalApiRpcLocalOptions { + std::function( + const DhtAddress& targetId, + RecursiveOperation operation, + const DhtAddress& excludedPeer)> + executeRecursiveOperation; + std::function>( + const DhtAddress& key, + const ::google::protobuf::Any& data, + const DhtAddress& creator)> + storeDataToDht; +}; + +class ExternalApiRpcLocal { +private: + ExternalApiRpcLocalOptions options; + +public: + explicit ExternalApiRpcLocal(ExternalApiRpcLocalOptions options) + : options(std::move(options)) {} + + folly::coro::Task externalFetchData( + ExternalFetchDataRequest request, DhtCallContext context) { + const auto sender = context.incomingSourceDescriptor.value(); + const auto result = co_await this->options.executeRecursiveOperation( + Identifiers::getDhtAddressFromRaw(DhtAddressRaw{request.key()}), + RecursiveOperation::FETCH_DATA, + Identifiers::getNodeIdFromPeerDescriptor(sender)); + ExternalFetchDataResponse response; + for (const auto& entry : result.dataEntries) { + *response.add_entries() = entry; + } + co_return response; + } + + folly::coro::Task externalStoreData( + ExternalStoreDataRequest request, DhtCallContext context) { + const auto sender = context.incomingSourceDescriptor.value(); + const auto storers = co_await this->options.storeDataToDht( + Identifiers::getDhtAddressFromRaw(DhtAddressRaw{request.key()}), + request.data(), + Identifiers::getNodeIdFromPeerDescriptor(sender)); + ExternalStoreDataResponse response; + for (const auto& storer : storers) { + *response.add_storers() = storer; + } + co_return response; + } + + folly::coro::Task + externalFindClosestNodes( + ExternalFindClosestNodesRequest request, DhtCallContext context) { + const auto sender = context.incomingSourceDescriptor.value(); + const auto result = co_await this->options.executeRecursiveOperation( + Identifiers::getDhtAddressFromRaw(DhtAddressRaw{request.nodeid()}), + RecursiveOperation::FIND_CLOSEST_NODES, + Identifiers::getNodeIdFromPeerDescriptor(sender)); + ExternalFindClosestNodesResponse response; + for (const auto& node : result.closestNodes) { + *response.add_closestnodes() = node; + } + co_return response; + } +}; + +} // namespace streamr::dht diff --git a/packages/streamr-dht/modules/dht/ExternalApiRpcRemote.cppm b/packages/streamr-dht/modules/dht/ExternalApiRpcRemote.cppm new file mode 100644 index 00000000..ae4de8f3 --- /dev/null +++ b/packages/streamr-dht/modules/dht/ExternalApiRpcRemote.cppm @@ -0,0 +1,123 @@ +// Module streamr.dht.ExternalApiRpcRemote +// Ported from packages/dht/src/dht/ExternalApiRpcRemote.ts (v103.8.0-rc.3). +// The client used to run recursive operations (fetch/store/find-closest) on a +// remote node "via peer" — i.e. through a connected entry point while the +// local node is still joining. Every call swallows its own error and returns +// an empty result, matching the TS try/catch. +module; + +#include +#include +#include +#include + +#include // IWYU pragma: keep + +export module streamr.dht.ExternalApiRpcRemote; + +import streamr.dht.protos; + +import streamr.utils.CoroutineHelper; +import streamr.logger.SLogger; +import streamr.dht.DhtRpcClient; +import streamr.dht.DhtCallContext; +import streamr.dht.Identifiers; +import streamr.dht.RpcRemote; + +// Hoisted from the former header (file scope, NOT exported). +using streamr::logger::SLogger; + +export namespace streamr::dht { + +using ::dht::DataEntry; +using ::dht::ExternalFetchDataRequest; +using ::dht::ExternalFindClosestNodesRequest; +using ::dht::ExternalStoreDataRequest; +using ::dht::PeerDescriptor; +using streamr::dht::contact::RpcRemote; +using streamr::dht::rpcprotocol::DhtCallContext; + +using ExternalApiRpcClient = ::dht::ExternalApiRpcClient; + +class ExternalApiRpcRemote : public RpcRemote { +private: + static constexpr std::chrono::milliseconds defaultTimeout{10000}; + +public: + ExternalApiRpcRemote( + PeerDescriptor localPeerDescriptor, // NOLINT + PeerDescriptor remotePeerDescriptor, + ExternalApiRpcClient client) + : RpcRemote( + std::move(localPeerDescriptor), + std::move(remotePeerDescriptor), + std::move(client)) {} + + folly::coro::Task> externalFetchData( + const DhtAddress& key) { + ExternalFetchDataRequest request; + request.set_key(Identifiers::getRawFromDhtAddress(key)); + auto options = this->formDhtRpcOptions(); + try { + const auto response = co_await this->getClient().externalFetchData( + std::move(request), std::move(options), defaultTimeout); + std::vector entries; + entries.reserve(static_cast(response.entries_size())); + for (const auto& entry : response.entries()) { + entries.push_back(entry); + } + co_return entries; + } catch (const std::exception& err) { + SLogger::debug( + "externalFetchData failed: " + std::string(err.what())); + co_return std::vector{}; + } + } + + folly::coro::Task> storeData( + const DhtAddress& key, const ::google::protobuf::Any& data) { + ExternalStoreDataRequest request; + request.set_key(Identifiers::getRawFromDhtAddress(key)); + *request.mutable_data() = data; + auto options = this->formDhtRpcOptions(); + try { + const auto response = co_await this->getClient().externalStoreData( + std::move(request), std::move(options), defaultTimeout); + std::vector storers; + storers.reserve(static_cast(response.storers_size())); + for (const auto& storer : response.storers()) { + storers.push_back(storer); + } + co_return storers; + } catch (const std::exception& err) { + SLogger::debug( + "externalStoreData failed: " + std::string(err.what())); + co_return std::vector{}; + } + } + + folly::coro::Task> externalFindClosestNode( + const DhtAddress& key) { + ExternalFindClosestNodesRequest request; + request.set_nodeid(Identifiers::getRawFromDhtAddress(key)); + auto options = this->formDhtRpcOptions(); + try { + const auto response = + co_await this->getClient().externalFindClosestNodes( + std::move(request), std::move(options), defaultTimeout); + std::vector closestNodes; + closestNodes.reserve( + static_cast(response.closestnodes_size())); + for (const auto& node : response.closestnodes()) { + closestNodes.push_back(node); + } + co_return closestNodes; + } catch (const std::exception& err) { + SLogger::debug( + "externalFindClosestNodes failed: " + std::string(err.what())); + co_return std::vector{}; + } + } +}; + +} // namespace streamr::dht diff --git a/packages/streamr-dht/test/integration/DhtNodeExternalApiTest.cpp b/packages/streamr-dht/test/integration/DhtNodeExternalApiTest.cpp new file mode 100644 index 00000000..25432c91 --- /dev/null +++ b/packages/streamr-dht/test/integration/DhtNodeExternalApiTest.cpp @@ -0,0 +1,120 @@ +// Ported from packages/dht/test/integration/DhtNodeExternalAPI.test.ts +// (v103.8.0-rc.3): a non-joined node uses another node as its gateway to the +// DHT via the external API RPCs (storeData / externalFetchData / +// externalFindClosestNodes). expectEqualData is adapted: the C++ mock entry +// carries no packed data payload (see TestUtils.createMockDataEntry), so +// entries are identified by their key. +#include +#include +#include + +#include // IWYU pragma: keep + +import streamr.utils.CoroutineHelper; +import streamr.dht.DhtNode; +import streamr.dht.Identifiers; +import streamr.dht.RegionPings; +import streamr.dht.Simulator; +import streamr.dht.SimulatorTransport; +import streamr.dht.TestUtils; +import streamr.dht.protos; + +#include "DhtNodeTestUtils.hpp" // IWYU pragma: keep — needs the imports above + +using ::dht::DataEntry; +using streamr::dht::DhtAddress; +using streamr::dht::Identifiers; +using streamr::dht::connection::simulator::LatencyType; +using streamr::dht::connection::simulator::Simulator; +using streamr::dht::testutils::createMockConnectionDhtNode; +using streamr::dht::testutils::createMockDataEntry; +using streamr::dht::testutils::MockDhtNode; +using streamr::utils::blockingWait; + +namespace { + +DhtAddress keyOf(const DataEntry& entry) { + return Identifiers::getDhtAddressFromRaw( + streamr::dht::DhtAddressRaw{entry.key()}); +} + +DhtAddress creatorOf(const DataEntry& entry) { + return Identifiers::getDhtAddressFromRaw( + streamr::dht::DhtAddressRaw{entry.creator()}); +} + +// expectEqualData(actual, expected): the TS helper compares the key and the +// unpacked data payload — NOT the creator (storeDataToDht stamps the storing +// node as creator, overriding the mock entry's random one). The C++ mock +// entry has no payload, so the key identifies the entry. +void expectEqualData(const DataEntry& actual, const DataEntry& expected) { + EXPECT_EQ(keyOf(actual), keyOf(expected)); +} + +} // namespace + +class DhtNodeExternalApiTest : public ::testing::Test { +protected: + Simulator simulator{LatencyType::NONE}; + std::shared_ptr dhtNode1; + std::shared_ptr remote; + + void SetUp() override { + this->dhtNode1 = createMockConnectionDhtNode(this->simulator); + this->remote = createMockConnectionDhtNode(this->simulator); + blockingWait(this->dhtNode1->node->joinDht( + {this->dhtNode1->node->getLocalPeerDescriptor()})); + } + + void TearDown() override { + this->dhtNode1->stopNode(); + this->remote->stopNode(); + this->simulator.stop(); + } +}; + +TEST_F(DhtNodeExternalApiTest, FetchDataHappyPath) { + const auto entry = createMockDataEntry(); + blockingWait( + this->dhtNode1->node->storeDataToDht(keyOf(entry), entry.data())); + const auto foundData = + blockingWait(this->remote->node->fetchDataFromDhtViaPeer( + keyOf(entry), this->dhtNode1->node->getLocalPeerDescriptor())); + ASSERT_FALSE(foundData.empty()); + expectEqualData(foundData[0], entry); +} + +TEST_F(DhtNodeExternalApiTest, FetchDataReturnsEmptyArrayIfNoDataFound) { + const auto foundData = + blockingWait(this->remote->node->fetchDataFromDhtViaPeer( + Identifiers::createRandomDhtAddress(), + this->dhtNode1->node->getLocalPeerDescriptor())); + EXPECT_TRUE(foundData.empty()); +} + +TEST_F(DhtNodeExternalApiTest, ExternalStoreDataHappyPath) { + const auto entry = createMockDataEntry(); + blockingWait(this->remote->node->storeDataToDhtViaPeer( + keyOf(entry), + entry.data(), + this->dhtNode1->node->getLocalPeerDescriptor())); + const auto foundData = + blockingWait(this->remote->node->fetchDataFromDhtViaPeer( + keyOf(entry), this->dhtNode1->node->getLocalPeerDescriptor())); + ASSERT_FALSE(foundData.empty()); + EXPECT_EQ(keyOf(foundData[0]), keyOf(entry)); + // The creator of externally stored data is the requesting node. + EXPECT_EQ( + creatorOf(foundData[0]), + Identifiers::getNodeIdFromPeerDescriptor( + this->remote->node->getLocalPeerDescriptor())); +} + +TEST_F(DhtNodeExternalApiTest, ExternalFindClosestNodesHappyPath) { + const auto nodeId = this->remote->node->getNodeId(); + const auto closestNodes = + blockingWait(this->remote->node->findClosestNodesFromDht(nodeId)); + ASSERT_FALSE(closestNodes.empty()); + EXPECT_EQ( + Identifiers::getNodeIdFromPeerDescriptor(closestNodes[0]), nodeId); +} diff --git a/packages/streamr-dht/test/integration/DhtNodeTest.cpp b/packages/streamr-dht/test/integration/DhtNodeTest.cpp new file mode 100644 index 00000000..6dd36771 --- /dev/null +++ b/packages/streamr-dht/test/integration/DhtNodeTest.cpp @@ -0,0 +1,186 @@ +// Ported from packages/dht/test/integration/DhtNode.test.ts (v103.8.0-rc.3): +// a DhtNode over a FakeTransport joins a DHT whose other members are RPC +// stubs (a ListeningRpcCommunicator + DhtNodeRpcLocal each, answering ping +// and getClosestPeers with the full membership), and ends up with every +// other member as a neighbour and closest contact. +#include +#include +#include +#include +#include +#include +#include + +#include // IWYU pragma: keep + +import streamr.utils.CoroutineHelper; +import streamr.utils.waitForCondition; +import streamr.dht.DhtCallContext; +import streamr.dht.DhtNode; +import streamr.dht.DhtNodeRpcLocal; +import streamr.dht.DhtNodeRpcRemote; +import streamr.dht.FakeTransport; +import streamr.dht.Identifiers; +import streamr.dht.ListeningRpcCommunicator; +import streamr.dht.ringIdentifiers; +import streamr.dht.TestUtils; +import streamr.dht.protos; + +using ::dht::ClosestPeersRequest; +using ::dht::ClosestPeersResponse; +using ::dht::PeerDescriptor; +using ::dht::PingRequest; +using ::dht::PingResponse; +using streamr::dht::ClosestRingPeerDescriptors; +using streamr::dht::DhtNode; +using streamr::dht::DhtNodeOptions; +using streamr::dht::DhtNodeRpcLocal; +using streamr::dht::DhtNodeRpcLocalOptions; +using streamr::dht::Identifiers; +using streamr::dht::ServiceID; +using streamr::dht::contact::RingIdRaw; +using streamr::dht::rpcprotocol::DhtCallContext; +using streamr::dht::testutils::createMockPeerDescriptor; +using streamr::dht::transport::FakeEnvironment; +using streamr::dht::transport::FakeTransport; +using streamr::dht::transport::ListeningRpcCommunicator; +using streamr::utils::blockingWait; +using streamr::utils::waitForCondition; + +namespace { + +constexpr size_t otherNodeCount = 3; +constexpr auto serviceIdLayer0 = "layer0"; +constexpr auto convergenceTimeout = std::chrono::seconds(10); +constexpr auto convergencePollInterval = std::chrono::milliseconds(100); + +// The transport and communicator of one RPC-stub member (TS keeps them alive +// via closure capture; here the fixture owns them). +struct RemoteStubNode { + std::shared_ptr transport; + std::unique_ptr rpcCommunicator; + std::unique_ptr dhtNodeRpcLocal; +}; + +} // namespace + +class DhtNodeTest : public ::testing::Test { +protected: + PeerDescriptor localPeerDescriptor; + PeerDescriptor entryPointPeerDescriptor; + std::vector otherPeerDescriptors; + std::vector> stubNodes; + + void SetUp() override { + this->localPeerDescriptor = createMockPeerDescriptor(); + this->entryPointPeerDescriptor = createMockPeerDescriptor(); + for (size_t i = 0; i < otherNodeCount; i++) { + this->otherPeerDescriptors.push_back(createMockPeerDescriptor()); + } + } + + [[nodiscard]] std::vector getAllPeerDescriptors() const { + std::vector all{ + this->localPeerDescriptor, this->entryPointPeerDescriptor}; + all.insert( + all.end(), + this->otherPeerDescriptors.begin(), + this->otherPeerDescriptors.end()); + return all; + } + + void startRemoteNode( + const PeerDescriptor& peerDescriptor, FakeEnvironment& environment) { + auto stub = std::make_unique(); + stub->transport = environment.createTransport(peerDescriptor); + stub->rpcCommunicator = std::make_unique( + ServiceID{serviceIdLayer0}, *stub->transport); + stub->dhtNodeRpcLocal = + std::make_unique(DhtNodeRpcLocalOptions{ + .peerDiscoveryQueryBatchSize = otherNodeCount + 2, + .getNeighbors = + [this, peerDescriptor]() { + // without(getAllPeerDescriptors(), peerDescriptor) + std::vector neighbors; + for (const auto& descriptor : + this->getAllPeerDescriptors()) { + if (!Identifiers::areEqualPeerDescriptors( + descriptor, peerDescriptor)) { + neighbors.push_back(descriptor); + } + } + return neighbors; + }, + // Never called in this test (no ring queries); TS passes + // undefined. + .getClosestRingContactsTo = + [](const RingIdRaw& /*ringIdRaw*/, size_t /*limit*/) { + return ClosestRingPeerDescriptors{}; + }, + .addContact = [](const PeerDescriptor& /*contact*/) {}, + // Never called in this test; TS passes undefined. + .removeContact = [](const streamr::dht::DhtAddress& + /*nodeId*/) {}}); + auto* rpcLocal = stub->dhtNodeRpcLocal.get(); + stub->rpcCommunicator->registerRpcMethod( + "ping", + [rpcLocal](const PingRequest& req, const DhtCallContext& context) { + return rpcLocal->ping(req, context); + }); + stub->rpcCommunicator + ->registerRpcMethod( + "getClosestPeers", + [rpcLocal]( + const ClosestPeersRequest& req, + const DhtCallContext& context) { + return rpcLocal->getClosestPeers(req, context); + }); + this->stubNodes.push_back(std::move(stub)); + } +}; + +TEST_F(DhtNodeTest, StartNodeAndJoinDht) { + FakeEnvironment environment; + this->startRemoteNode(this->entryPointPeerDescriptor, environment); + for (const auto& other : this->otherPeerDescriptors) { + this->startRemoteNode(other, environment); + } + + auto transport = environment.createTransport(this->localPeerDescriptor); + DhtNode localNode( + DhtNodeOptions{ + .transport = transport.get(), + .connectionsView = transport.get(), + .peerDescriptor = this->localPeerDescriptor, + .entryPoints = {this->entryPointPeerDescriptor}}); + blockingWait(localNode.start()); + blockingWait(localNode.joinDht({this->entryPointPeerDescriptor})); + blockingWait(localNode.waitForNetworkConnectivity()); + + blockingWait(waitForCondition( + [&localNode]() { + return localNode.getNeighborCount() == otherNodeCount + 1; + }, + convergenceTimeout, + convergencePollInterval)); + + // toIncludeSameMembers(expected, actual) on node ids. + std::set expectedNodeIds; + for (const auto& descriptor : this->getAllPeerDescriptors()) { + if (!Identifiers::areEqualPeerDescriptors( + descriptor, this->localPeerDescriptor)) { + expectedNodeIds.insert( + static_cast( + Identifiers::getNodeIdFromPeerDescriptor(descriptor))); + } + } + std::set actualNodeIds; + for (const auto& descriptor : localNode.getClosestContacts()) { + actualNodeIds.insert( + static_cast( + Identifiers::getNodeIdFromPeerDescriptor(descriptor))); + } + EXPECT_EQ(actualNodeIds, expectedNodeIds); + + localNode.stop(); +} diff --git a/packages/streamr-dht/test/integration/KademliaCorrectnessTest.cpp b/packages/streamr-dht/test/integration/KademliaCorrectnessTest.cpp new file mode 100644 index 00000000..4ae18085 --- /dev/null +++ b/packages/streamr-dht/test/integration/KademliaCorrectnessTest.cpp @@ -0,0 +1,172 @@ +// Ported from packages/dht/test/benchmark/KademliaCorrectness.test.ts +// (v103.8.0-rc.3), adapted per the completion plan ("as a slow correctness +// test, not a benchmark"): +// - the TS test reads pre-generated ground-truth files (test/data/*.json, +// produced by `npm run prepare-kademlia-simulation` and not in the repo); +// the ground truth is simply each node's neighbours ordered by Kademlia +// XOR distance, so it is computed in-test with getClosestNodes over the +// actual node ids instead of loaded from files (which also lets the ids +// be random per run); +// - the TS test only prints the statistics; here they are asserted with +// conservative floors so regressions in join/discovery correctness fail +// the test. +// Like TS, a node's "correct neighbours" is the length of the matching +// PREFIX between its getClosestContacts(8) ordering and the ground-truth +// ordering. +#include +#include +#include +#include +#include +#include + +#include // IWYU pragma: keep + +import streamr.logger.SLogger; +import streamr.utils.CoroutineHelper; +import streamr.dht.DhtNode; +import streamr.dht.getClosestNodes; +import streamr.dht.Identifiers; +import streamr.dht.RegionPings; +import streamr.dht.Simulator; +import streamr.dht.SimulatorTransport; +import streamr.dht.protos; + +#include "DhtNodeTestUtils.hpp" // IWYU pragma: keep — needs the imports above + +using ::dht::PeerDescriptor; +using streamr::dht::DhtAddress; +using streamr::dht::Identifiers; +using streamr::dht::connection::simulator::Simulator; +using streamr::dht::contact::getClosestNodes; +using streamr::dht::contact::GetClosestNodesOptions; +using streamr::dht::testutils::createMockConnectionDhtNode; +using streamr::dht::testutils::MockDhtNode; +using streamr::utils::blockingWait; + +namespace { + +constexpr size_t numNodes = 200; +constexpr size_t neighborsToCheck = 8; + +} // namespace + +class KademliaCorrectnessTest : public ::testing::Test { +protected: + Simulator simulator; + std::vector> nodes; + + void SetUp() override { + for (size_t i = 0; i < numNodes; i++) { + this->nodes.push_back(createMockConnectionDhtNode(this->simulator)); + } + } + + void TearDown() override { + for (const auto& node : this->nodes) { + node->stopNode(); + } + this->simulator.stop(); + } +}; + +TEST_F(KademliaCorrectnessTest, CanFindCorrectNeighbors) { + const auto& entryPoint = this->nodes[0]; + const auto entryPointDescriptor = + entryPoint->node->getLocalPeerDescriptor(); + blockingWait(entryPoint->node->joinDht({entryPointDescriptor})); + for (size_t i = 1; i < numNodes; i++) { + blockingWait(this->nodes[i]->node->joinDht({entryPointDescriptor})); + } + + // Ground truth: for each node, every other node's descriptor ordered by + // Kademlia XOR distance from it (what the TS pre-generated data holds). + std::vector allDescriptors; + allDescriptors.reserve(numNodes); + for (const auto& node : this->nodes) { + allDescriptors.push_back(node->node->getLocalPeerDescriptor()); + } + + size_t minimumCorrectNeighbors = numNodes; + size_t sumCorrectNeighbors = 0; + size_t sumKbucketSize = 1; + size_t sumKnownOfClosest = 0; + + for (size_t i = 0; i < numNodes; i++) { + const auto nodeId = this->nodes[i]->node->getNodeId(); + std::vector others; + others.reserve(numNodes - 1); + for (size_t j = 0; j < numNodes; j++) { + if (j != i) { + others.push_back(allDescriptors[j]); + } + } + const auto groundTruth = getClosestNodes( + nodeId, + others, + GetClosestNodesOptions{.maxCount = neighborsToCheck}); + + const auto kademliaNeighbors = + this->nodes[i]->node->getClosestContacts(neighborsToCheck); + + size_t correctNeighbors = 0; + for (size_t j = 0; + j < groundTruth.size() && j < kademliaNeighbors.size(); + j++) { + if (!Identifiers::areEqualPeerDescriptors( + groundTruth[j], kademliaNeighbors[j])) { + break; + } + correctNeighbors++; + } + minimumCorrectNeighbors = + std::min(minimumCorrectNeighbors, correctNeighbors); + // Order-insensitive complement of the strict prefix metric above: + // how many of the true closest-8 the node knows AT ALL (anywhere in + // its top-8). Separates "does not know its closest peers" (a real + // discovery failure) from "knows them in slightly different order" + // (inherent to Kademlia's liveness-biased buckets). + size_t knownOfClosest = 0; + for (const auto& truth : groundTruth) { + for (const auto& neighbor : kademliaNeighbors) { + if (Identifiers::areEqualPeerDescriptors(truth, neighbor)) { + knownOfClosest++; + break; + } + } + } + if (i > 0) { + sumKbucketSize += this->nodes[i]->node->getNeighborCount(); + sumCorrectNeighbors += correctNeighbors; + sumKnownOfClosest += knownOfClosest; + } + } + + const double avgKbucketSize = + static_cast(sumKbucketSize) / (numNodes - 1); + const double avgCorrectNeighbors = + static_cast(sumCorrectNeighbors) / (numNodes - 1); + const double avgKnownOfClosest = + static_cast(sumKnownOfClosest) / (numNodes - 1); + + // TS prints these; keep them visible in the test log. + streamr::logger::SLogger::info( + "----------- Simulation results ------------------"); + streamr::logger::SLogger::info( + "Minimum correct neighbors: " + + std::to_string(minimumCorrectNeighbors)); + streamr::logger::SLogger::info( + "Average correct neighbors: " + std::to_string(avgCorrectNeighbors)); + streamr::logger::SLogger::info( + "Average Kbucket size: " + std::to_string(avgKbucketSize)); + streamr::logger::SLogger::info( + "Average known-of-closest-8 (order-insensitive): " + + std::to_string(avgKnownOfClosest)); + + // Adaptation: conservative correctness floors (the TS test asserts + // nothing). A healthy 200-node Kademlia converges most nodes to their + // exact closest-neighbour prefix; a broken join/discovery collapses + // these numbers to near zero. + EXPECT_GE(avgCorrectNeighbors, 4.0); + EXPECT_GE(avgKbucketSize, static_cast(neighborsToCheck) / 2); +} diff --git a/packages/streamr-dht/test/integration/Layer1ScaleTest.cpp b/packages/streamr-dht/test/integration/Layer1ScaleTest.cpp new file mode 100644 index 00000000..3157a61f --- /dev/null +++ b/packages/streamr-dht/test/integration/Layer1ScaleTest.cpp @@ -0,0 +1,165 @@ +// Ported from packages/dht/test/integration/Layer1-scale.test.ts +// (v103.8.0-rc.3): 48 layer-0 DhtNodes over the Simulator join in parallel; +// then layer-1 DhtNodes stacked on them (one service, and four services in +// the second test) join in parallel, and every layer-1 node must see exactly +// its layer-0 node's connections (the layers share one physical connection +// set). The commented-out TS 'layer1 routing' case is not ported (disabled +// upstream: "TODO: Make this work"). +#include +#include +#include +#include +#include + +#include // IWYU pragma: keep + +import streamr.utils.CoroutineHelper; +import streamr.dht.DhtNode; +import streamr.dht.Identifiers; +import streamr.dht.RegionPings; +import streamr.dht.Simulator; +import streamr.dht.SimulatorTransport; +import streamr.dht.TestUtils; +import streamr.dht.protos; + +#include "DhtNodeTestUtils.hpp" // IWYU pragma: keep — needs the imports above + +using ::dht::PeerDescriptor; +using streamr::dht::Identifiers; +using streamr::dht::numberOfNodesPerKBucketDefault; +using streamr::dht::connection::simulator::Simulator; +using streamr::dht::testutils::createMockConnectionDhtNode; +using streamr::dht::testutils::createMockConnectionLayer1Node; +using streamr::dht::testutils::createMockPeerDescriptor; +using streamr::dht::testutils::MockDhtNode; +using streamr::utils::blockingWait; + +namespace { + +constexpr size_t nodeCount = 48; + +std::vector toNodeIds( + const std::vector& descriptors) { + std::vector ids; + ids.reserve(descriptors.size()); + for (const auto& descriptor : descriptors) { + ids.push_back( + static_cast( + Identifiers::getNodeIdFromPeerDescriptor(descriptor))); + } + return ids; +} + +// Promise.all(nodes.map((node) => node.joinDht([entryPoint]))) +void joinAllInParallel( + const std::vector>& nodes, + const PeerDescriptor& entryPointDescriptor) { + std::vector> joins; + joins.reserve(nodes.size()); + for (const auto& node : nodes) { + joins.push_back(node->node->joinDht({entryPointDescriptor})); + } + blockingWait(folly::coro::collectAllRange(std::move(joins))); +} + +} // namespace + +class Layer1ScaleTest : public ::testing::Test { +protected: + Simulator simulator; + PeerDescriptor entryPoint0Descriptor = createMockPeerDescriptor(); + std::shared_ptr layer0EntryPoint; + std::vector> nodes; + std::vector> layer1CleanUp; + + void SetUp() override { + this->layer0EntryPoint = createMockConnectionDhtNode( + this->simulator, + Identifiers::getNodeIdFromPeerDescriptor( + this->entryPoint0Descriptor)); + blockingWait(this->layer0EntryPoint->node->joinDht( + {this->entryPoint0Descriptor})); + + for (size_t i = 0; i < nodeCount; i++) { + this->nodes.push_back(createMockConnectionDhtNode(this->simulator)); + } + joinAllInParallel(this->nodes, this->entryPoint0Descriptor); + } + + void TearDown() override { + for (const auto& node : this->layer1CleanUp) { + node->stopNode(); + } + for (const auto& node : this->nodes) { + node->stopNode(); + } + this->layer0EntryPoint->stopNode(); + this->simulator.stop(); + } +}; + +TEST_F(Layer1ScaleTest, SingleLayer1Dht) { + auto layer1EntryPoint = + createMockConnectionLayer1Node(*this->layer0EntryPoint->node); + blockingWait( + layer1EntryPoint->node->joinDht({this->entryPoint0Descriptor})); + this->layer1CleanUp.push_back(layer1EntryPoint); + + std::vector> layer1Nodes; + for (size_t i = 0; i < nodeCount; i++) { + auto layer1 = createMockConnectionLayer1Node(*this->nodes[i]->node); + layer1Nodes.push_back(layer1); + this->layer1CleanUp.push_back(layer1); + } + + joinAllInParallel(layer1Nodes, this->entryPoint0Descriptor); + + for (size_t i = 0; i < nodeCount; i++) { + const auto& layer0Node = this->nodes[i]->node; + const auto& layer1Node = layer1Nodes[i]->node; + EXPECT_EQ(layer1Node->getNodeId(), layer0Node->getNodeId()); + EXPECT_EQ( + layer1Node->getConnectionsView()->getConnectionCount(), + layer0Node->getConnectionsView()->getConnectionCount()); + EXPECT_GE( + layer1Node->getNeighborCount(), numberOfNodesPerKBucketDefault / 2); + EXPECT_EQ( + toNodeIds(layer1Node->getConnectionsView()->getConnections()), + toNodeIds(layer0Node->getConnectionsView()->getConnections())); + } +} + +TEST_F(Layer1ScaleTest, MultipleLayer1Dht) { + const std::vector serviceIds = {"one", "two", "three", "four"}; + for (const auto& serviceId : serviceIds) { + auto entryPoint = createMockConnectionLayer1Node( + *this->layer0EntryPoint->node, serviceId); + blockingWait(entryPoint->node->joinDht({this->entryPoint0Descriptor})); + this->layer1CleanUp.push_back(entryPoint); + } + + std::vector>> streams( + serviceIds.size()); + for (size_t i = 0; i < nodeCount; i++) { + for (size_t s = 0; s < serviceIds.size(); s++) { + auto layer1 = createMockConnectionLayer1Node( + *this->nodes[i]->node, serviceIds[s]); + streams[s].push_back(layer1); + this->layer1CleanUp.push_back(layer1); + } + } + + // TS joins layer1CleanUp (the already-joined entry points re-join too). + joinAllInParallel(this->layer1CleanUp, this->entryPoint0Descriptor); + + for (size_t i = 0; i < nodeCount; i++) { + const auto& layer0Node = this->nodes[i]->node; + for (size_t s = 0; s < serviceIds.size(); s++) { + EXPECT_EQ( + layer0Node->getConnectionsView()->getConnectionCount(), + streams[s][i] + ->node->getConnectionsView() + ->getConnectionCount()); + } + } +} diff --git a/packages/streamr-dht/test/integration/MockLayer1Layer0Test.cpp b/packages/streamr-dht/test/integration/MockLayer1Layer0Test.cpp new file mode 100644 index 00000000..0f9ae3ff --- /dev/null +++ b/packages/streamr-dht/test/integration/MockLayer1Layer0Test.cpp @@ -0,0 +1,147 @@ +// Ported from packages/dht/test/integration/Mock-Layer1-Layer0.test.ts +// (v103.8.0-rc.3): five layer-0 DhtNodes over the Simulator, each carrying a +// layer-1 DhtNode that uses the layer-0 node as its transport. After all +// nodes join both layers, each layer-1 node must converge on the same +// neighbour set as its layer-0 node (the layers share the physical +// connections, so the topologies coincide in this small network). +#include +#include +#include +#include +#include + +#include // IWYU pragma: keep + +import streamr.utils.CoroutineHelper; +import streamr.dht.DhtNode; +import streamr.dht.Identifiers; +import streamr.dht.RegionPings; +import streamr.dht.Simulator; +import streamr.dht.SimulatorTransport; +import streamr.dht.protos; + +#include "DhtNodeTestUtils.hpp" // IWYU pragma: keep — needs the imports above + +using ::dht::PeerDescriptor; +using streamr::dht::Identifiers; +using streamr::dht::connection::simulator::Simulator; +using streamr::dht::testutils::createMockConnectionDhtNode; +using streamr::dht::testutils::createMockConnectionLayer1Node; +using streamr::dht::testutils::MockDhtNode; +using streamr::utils::blockingWait; + +namespace { + +std::set toNodeIdSet( + const std::vector& descriptors) { + std::set ids; + for (const auto& descriptor : descriptors) { + ids.insert( + static_cast( + Identifiers::getNodeIdFromPeerDescriptor(descriptor))); + } + return ids; +} + +} // namespace + +class MockLayer1Layer0Test : public ::testing::Test { +protected: + Simulator simulator; + std::shared_ptr layer0EntryPoint; + std::shared_ptr layer0Node1; + std::shared_ptr layer0Node2; + std::shared_ptr layer0Node3; + std::shared_ptr layer0Node4; + std::shared_ptr layer1EntryPoint; + std::shared_ptr layer1Node1; + std::shared_ptr layer1Node2; + std::shared_ptr layer1Node3; + std::shared_ptr layer1Node4; + + void SetUp() override { + this->layer0EntryPoint = createMockConnectionDhtNode( + this->simulator, Identifiers::createRandomDhtAddress()); + this->layer0Node1 = createMockConnectionDhtNode( + this->simulator, Identifiers::createRandomDhtAddress()); + this->layer0Node2 = createMockConnectionDhtNode( + this->simulator, Identifiers::createRandomDhtAddress()); + this->layer0Node3 = createMockConnectionDhtNode( + this->simulator, Identifiers::createRandomDhtAddress()); + this->layer0Node4 = createMockConnectionDhtNode( + this->simulator, Identifiers::createRandomDhtAddress()); + + this->layer1EntryPoint = + createMockConnectionLayer1Node(*this->layer0EntryPoint->node); + this->layer1Node1 = + createMockConnectionLayer1Node(*this->layer0Node1->node); + this->layer1Node2 = + createMockConnectionLayer1Node(*this->layer0Node2->node); + this->layer1Node3 = + createMockConnectionLayer1Node(*this->layer0Node3->node); + this->layer1Node4 = + createMockConnectionLayer1Node(*this->layer0Node4->node); + + const auto entryPointDescriptor = + this->layer0EntryPoint->node->getLocalPeerDescriptor(); + blockingWait( + this->layer0EntryPoint->node->joinDht({entryPointDescriptor})); + blockingWait( + this->layer1EntryPoint->node->joinDht({entryPointDescriptor})); + } + + void TearDown() override { + // Layer-1 nodes first: they use the layer-0 nodes as transports. + this->layer1EntryPoint->stopNode(); + this->layer1Node1->stopNode(); + this->layer1Node2->stopNode(); + this->layer1Node3->stopNode(); + this->layer1Node4->stopNode(); + this->layer0EntryPoint->stopNode(); + this->layer0Node1->stopNode(); + this->layer0Node2->stopNode(); + this->layer0Node3->stopNode(); + this->layer0Node4->stopNode(); + this->simulator.stop(); + } +}; + +TEST_F(MockLayer1Layer0Test, HappyPath) { + const auto entryPointDescriptor = + this->layer0EntryPoint->node->getLocalPeerDescriptor(); + blockingWait(this->layer0Node1->node->joinDht({entryPointDescriptor})); + blockingWait(this->layer0Node2->node->joinDht({entryPointDescriptor})); + blockingWait(this->layer0Node3->node->joinDht({entryPointDescriptor})); + blockingWait(this->layer0Node4->node->joinDht({entryPointDescriptor})); + + blockingWait(this->layer1Node1->node->joinDht({entryPointDescriptor})); + blockingWait(this->layer1Node2->node->joinDht({entryPointDescriptor})); + blockingWait(this->layer1Node3->node->joinDht({entryPointDescriptor})); + blockingWait(this->layer1Node4->node->joinDht({entryPointDescriptor})); + + EXPECT_EQ( + this->layer1Node1->node->getNeighborCount(), + this->layer0Node1->node->getNeighborCount()); + EXPECT_EQ( + this->layer1Node2->node->getNeighborCount(), + this->layer0Node2->node->getNeighborCount()); + EXPECT_EQ( + this->layer1Node3->node->getNeighborCount(), + this->layer0Node3->node->getNeighborCount()); + EXPECT_EQ( + this->layer1Node4->node->getNeighborCount(), + this->layer0Node4->node->getNeighborCount()); + + EXPECT_EQ( + toNodeIdSet(this->layer1Node1->node->getNeighbors()), + toNodeIdSet(this->layer0Node1->node->getNeighbors())); + EXPECT_EQ( + toNodeIdSet(this->layer1Node2->node->getNeighbors()), + toNodeIdSet(this->layer0Node2->node->getNeighbors())); + EXPECT_EQ( + toNodeIdSet(this->layer1Node3->node->getNeighbors()), + toNodeIdSet(this->layer0Node3->node->getNeighbors())); + EXPECT_EQ( + toNodeIdSet(this->layer1Node4->node->getNeighbors()), + toNodeIdSet(this->layer0Node4->node->getNeighbors())); +} diff --git a/packages/streamr-dht/test/integration/MultipleEntryPointJoiningTest.cpp b/packages/streamr-dht/test/integration/MultipleEntryPointJoiningTest.cpp new file mode 100644 index 00000000..2c0d0077 --- /dev/null +++ b/packages/streamr-dht/test/integration/MultipleEntryPointJoiningTest.cpp @@ -0,0 +1,119 @@ +// Ported from packages/dht/test/integration/MultipleEntryPointJoining.test.ts +// (v103.8.0-rc.3): DhtNodes over the Simulator joining through several entry +// points at once (deferred here from phase A7). Exercises the whole node — +// PeerDiscovery, Router, RecursiveOperationManager and the RPC stack — over +// real simulated connections. +#include +#include +#include +#include +#include + +#include // IWYU pragma: keep + +import streamr.utils.CoroutineHelper; +import streamr.dht.DhtNode; +import streamr.dht.Identifiers; +import streamr.dht.RegionPings; +import streamr.dht.Simulator; +import streamr.dht.SimulatorTransport; +import streamr.dht.protos; + +#include "DhtNodeTestUtils.hpp" // IWYU pragma: keep — needs the imports above + +using ::dht::PeerDescriptor; +using streamr::dht::connection::simulator::LatencyType; +using streamr::dht::connection::simulator::Simulator; +using streamr::dht::testutils::createMockConnectionDhtNode; +using streamr::dht::testutils::MockDhtNode; +using streamr::utils::blockingWait; + +class MultipleEntryPointJoiningTest : public ::testing::Test { +protected: + Simulator simulator{LatencyType::REAL}; + std::shared_ptr node1; + std::shared_ptr node2; + std::shared_ptr node3; + std::vector entryPoints; + + void SetUp() override { + this->node1 = createMockConnectionDhtNode(this->simulator); + this->node2 = createMockConnectionDhtNode(this->simulator); + this->node3 = createMockConnectionDhtNode(this->simulator); + this->entryPoints = { + this->node1->node->getLocalPeerDescriptor(), + this->node2->node->getLocalPeerDescriptor(), + this->node3->node->getLocalPeerDescriptor()}; + } + + void TearDown() override { + this->node1->stopNode(); + this->node2->stopNode(); + this->node3->stopNode(); + this->simulator.stop(); + } +}; + +TEST_F(MultipleEntryPointJoiningTest, CanJoinSimultaneously) { + blockingWait( + folly::coro::collectAll( + this->node1->node->joinDht(this->entryPoints), + this->node2->node->joinDht(this->entryPoints), + this->node3->node->joinDht(this->entryPoints))); + EXPECT_EQ(this->node1->node->getNeighborCount(), 2U); + EXPECT_EQ(this->node2->node->getNeighborCount(), 2U); + EXPECT_EQ(this->node3->node->getNeighborCount(), 2U); +} + +TEST_F(MultipleEntryPointJoiningTest, CanJoinEvenIfANodeIsOffline) { + this->node3->stopNode(); + blockingWait( + folly::coro::collectAll( + this->node1->node->joinDht(this->entryPoints), + this->node2->node->joinDht(this->entryPoints))); + EXPECT_EQ(this->node1->node->getNeighborCount(), 1U); + EXPECT_EQ(this->node2->node->getNeighborCount(), 1U); +} + +class JoinViaMultipleEntryPointsTest : public ::testing::Test { +protected: + Simulator simulator{LatencyType::REAL}; + std::shared_ptr entryPoint1; + std::shared_ptr entryPoint2; + std::shared_ptr node1; + std::shared_ptr node2; + std::vector entryPoints; + + void SetUp() override { + this->entryPoint1 = createMockConnectionDhtNode(this->simulator); + this->entryPoint2 = createMockConnectionDhtNode(this->simulator); + this->node1 = createMockConnectionDhtNode(this->simulator); + this->node2 = createMockConnectionDhtNode(this->simulator); + this->entryPoints = { + this->entryPoint1->node->getLocalPeerDescriptor(), + this->entryPoint2->node->getLocalPeerDescriptor()}; + blockingWait(this->entryPoint1->node->joinDht(this->entryPoints)); + blockingWait(this->entryPoint2->node->joinDht(this->entryPoints)); + } + + void TearDown() override { + this->entryPoint1->stopNode(); + this->entryPoint2->stopNode(); + this->node1->stopNode(); + this->node2->stopNode(); + this->simulator.stop(); + } +}; + +// Regression test for the phase-AA concurrent-connect defects (see +// trackerless-network-completion-plan.md, "Phase AA"): the joining node's +// k-bucket ping and its discovery query race to connect the same new peer on +// different threads. Fixed by (1) send() no longer closing the connector-shared +// pending connection on a tie-break rejection and (2) ConnectingEndpointState +// preserving message boundaries when flushing its buffer. +TEST_F(JoinViaMultipleEntryPointsTest, NonEntryPointNodesCanJoin) { + blockingWait(this->node1->node->joinDht(this->entryPoints)); + EXPECT_EQ(this->node1->node->getNeighborCount(), 2U); + blockingWait(this->node2->node->joinDht(this->entryPoints)); + EXPECT_EQ(this->node2->node->getNeighborCount(), 3U); +} diff --git a/packages/streamr-dht/test/utils/DhtNodeTestUtils.hpp b/packages/streamr-dht/test/utils/DhtNodeTestUtils.hpp new file mode 100644 index 00000000..edcf378a --- /dev/null +++ b/packages/streamr-dht/test/utils/DhtNodeTestUtils.hpp @@ -0,0 +1,82 @@ +// Ported from packages/dht/test/utils/utils.ts (v103.8.0-rc.3): helpers that +// build a DhtNode over a SimulatorTransport, and a layer-1 DhtNode over a +// layer-0 DhtNode. +// +// This is a plain header, NOT a module: a module that imports DhtNode's +// aggregated interface exhausts clang's source-location space during BMI +// generation ("ran out of source locations"). A .cpp that imports the modules +// and includes this header compiles to a .o without that BMI step. The +// including translation unit must therefore FIRST import: +// streamr.utils.CoroutineHelper, streamr.dht.DhtNode, +// streamr.dht.Identifiers, streamr.dht.protos, streamr.dht.RegionPings, +// streamr.dht.Simulator, streamr.dht.SimulatorTransport +// TS subclasses DhtNode to override stop() to also stop the mock transport; +// here the returned MockDhtNode owns the transport and stopNode() stops both +// (a layer-1 node shares the layer-0 node's transport, so it has none). +#pragma once + +#include +#include +#include +#include +#include + +namespace streamr::dht::testutils { + +struct MockDhtNode { + std::shared_ptr + transport; // null for a layer-1 node + std::shared_ptr node; + + void stopNode() { + this->node->stop(); + if (this->transport != nullptr) { + this->transport->stop(); + } + } +}; + +inline std::shared_ptr createMockConnectionDhtNode( + streamr::dht::connection::simulator::Simulator& simulator, + std::optional nodeId = std::nullopt) { + ::dht::PeerDescriptor peerDescriptor; + peerDescriptor.set_nodeid( + streamr::dht::Identifiers::getRawFromDhtAddress(nodeId.value_or( + streamr::dht::Identifiers::createRandomDhtAddress()))); + peerDescriptor.set_type(::dht::NodeType::NODEJS); + peerDescriptor.set_region( + static_cast( + streamr::dht::connection::simulator::getRandomRegion())); + auto transport = std::make_shared< + streamr::dht::connection::simulator::SimulatorTransport>( + peerDescriptor, simulator); + transport->start(); + auto node = + std::make_shared(streamr::dht::DhtNodeOptions{ + .rpcRequestTimeout = std::chrono::milliseconds(5000), + .transport = transport.get(), + .connectionsView = transport.get(), + .connectionLocker = transport.get(), + .peerDescriptor = peerDescriptor}); + streamr::utils::blockingWait(node->start()); + return std::make_shared(MockDhtNode{transport, node}); +} + +inline std::shared_ptr createMockConnectionLayer1Node( + streamr::dht::DhtNode& layer0Node, + const std::string& serviceId = "layer1") { + ::dht::PeerDescriptor descriptor; + descriptor.set_nodeid(layer0Node.getLocalPeerDescriptor().nodeid()); + descriptor.set_type(::dht::NodeType::NODEJS); + auto node = + std::make_shared(streamr::dht::DhtNodeOptions{ + .serviceId = streamr::dht::ServiceID{serviceId}, + .rpcRequestTimeout = std::chrono::milliseconds(10000), + .transport = &layer0Node, + .connectionsView = layer0Node.getConnectionsView(), + .peerDescriptor = descriptor}); + streamr::utils::blockingWait(node->start()); + return std::make_shared(MockDhtNode{nullptr, node}); +} + +} // namespace streamr::dht::testutils