From 10a73883149cb66795a8e830e3c65f3fb587f2dc Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Sun, 12 Jul 2026 01:31:08 +0300 Subject: [PATCH 1/3] Phase C4: entry points in the DHT (PeerDescriptorStoreManager, StreamPartNetworkSplitAvoidance, StreamPartReconnect) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ports from the pinned TS 103.8.0-rc.3 (af966cf03): - DiscoveryLayerNode (modules/discovery-layer/): the control layer's view of a stream part's layer-1 DHT node, ported in full (events included) so C5/C6 can reuse it; TS RingContacts maps to the dht package's ClosestRingPeerDescriptors. MockDiscoveryLayerNode joins the test-utils module library (mutex-guarded: C++ tests mutate the k-bucket from the test thread while detached loops read it on pool threads). - PeerDescriptorStoreManager (modules/control-layer/): fetch/store/keep stream entry points under a DHT key via callback-injected DHT operations; entry points virtual so StreamPartReconnect's test can substitute the TS fake (which force-casts an unrelated class). - StreamPartNetworkSplitAvoidance: exponential-run-off rejoin. The TS run-off runs ALL attempts even after the task stops throwing — the unit test pins that (expects the neighbor count to exceed, not reach, the minimum) — preserved. discoverEntryPoints is a parameterless callback (TS declares an optional excluded-nodes parameter it never passes). - StreamPartReconnect: periodic entry-point refetch + rejoin until a neighbor appears. Design deviation (C++ wins, per plan §2.2): TS detaches the interval loops (scheduleAtInterval) and relies on GC; a detached loop touching `this` is a use-after-free once the owner dies (see the Layer0 teardown SIGSEGV fixes, PR #81). The loops are owned by a per-instance CancellableAsyncScope drained in destroy() — awaited, never a blocking join on a pool thread (the sendResponse deadlock lesson) — with a blocking destructor backstop for owner threads. The phase's three integration tests need NetworkNode/createNetworkNode and move to milestone C6/C8 (noted in the plan). unit/StreamPartReconnect.test.ts exists in TS though the plan did not list it; ported too. Tests: 6 new, package suite 28/28. Co-Authored-By: Claude Fable 5 --- .../CMakeLists.txt | 6 +- .../StreamPartNetworkSplitAvoidance.cppm | 166 +++++++++++++ .../modules/StreamPartReconnect.cppm | 150 ++++++++++++ .../PeerDescriptorStoreManager.cppm | 227 ++++++++++++++++++ .../discovery-layer/DiscoveryLayerNode.cppm | 78 ++++++ .../unit/PeerDescriptorStoreManagerTest.cpp | 122 ++++++++++ .../StreamPartNetworkSplitAvoidanceTest.cpp | 54 +++++ .../test/unit/StreamPartReconnectTest.cpp | 101 ++++++++ .../test/utils/MockDiscoveryLayerNode.cppm | 97 ++++++++ trackerless-network-completion-plan.md | 13 + 10 files changed, 1013 insertions(+), 1 deletion(-) create mode 100644 packages/streamr-trackerless-network/modules/StreamPartNetworkSplitAvoidance.cppm create mode 100644 packages/streamr-trackerless-network/modules/StreamPartReconnect.cppm create mode 100644 packages/streamr-trackerless-network/modules/control-layer/PeerDescriptorStoreManager.cppm create mode 100644 packages/streamr-trackerless-network/modules/discovery-layer/DiscoveryLayerNode.cppm create mode 100644 packages/streamr-trackerless-network/test/unit/PeerDescriptorStoreManagerTest.cpp create mode 100644 packages/streamr-trackerless-network/test/unit/StreamPartNetworkSplitAvoidanceTest.cpp create mode 100644 packages/streamr-trackerless-network/test/unit/StreamPartReconnectTest.cpp create mode 100644 packages/streamr-trackerless-network/test/utils/MockDiscoveryLayerNode.cppm diff --git a/packages/streamr-trackerless-network/CMakeLists.txt b/packages/streamr-trackerless-network/CMakeLists.txt index 9351d434..d20045c2 100644 --- a/packages/streamr-trackerless-network/CMakeLists.txt +++ b/packages/streamr-trackerless-network/CMakeLists.txt @@ -143,7 +143,8 @@ if(NOT IOS AND STREAMR_MODULES_SUPPORTED) # import streamr.trackerlessnetwork.TestUtils. streamr_add_module_library(streamr-trackerless-network-test-utils BASE_DIRS ${CMAKE_CURRENT_SOURCE_DIR}/test/utils - FILES ${CMAKE_CURRENT_SOURCE_DIR}/test/utils/TestUtils.cppm) + FILES ${CMAKE_CURRENT_SOURCE_DIR}/test/utils/TestUtils.cppm + ${CMAKE_CURRENT_SOURCE_DIR}/test/utils/MockDiscoveryLayerNode.cppm) target_include_directories(streamr-trackerless-network-test-utils PUBLIC $) target_link_libraries(streamr-trackerless-network-test-utils @@ -154,6 +155,9 @@ if(NOT IOS AND STREAMR_MODULES_SUPPORTED) add_executable(streamr-trackerless-network-test-unit test/unit/TestUtilsTest.cpp + test/unit/PeerDescriptorStoreManagerTest.cpp + test/unit/StreamPartNetworkSplitAvoidanceTest.cpp + test/unit/StreamPartReconnectTest.cpp test/unit/ProxyClientTest.cpp test/unit/ProxyConnectionRpcLocalTest.cpp test/unit/ProxyConnectionRpcRemoteTest.cpp diff --git a/packages/streamr-trackerless-network/modules/StreamPartNetworkSplitAvoidance.cppm b/packages/streamr-trackerless-network/modules/StreamPartNetworkSplitAvoidance.cppm new file mode 100644 index 00000000..37f87e16 --- /dev/null +++ b/packages/streamr-trackerless-network/modules/StreamPartNetworkSplitAvoidance.cppm @@ -0,0 +1,166 @@ +// Module streamr.trackerlessnetwork.StreamPartNetworkSplitAvoidance +// Ported from packages/trackerless-network/src/ +// StreamPartNetworkSplitAvoidance.ts (v103.8.0-rc.3): tries to find new +// neighbors when the node has fewer than MIN_NEIGHBOR_COUNT by rejoining +// the stream's control-layer network, avoiding some network-split +// scenarios (most relevant for small stream networks). +// +// Port notes: TS exponentialRunOff runs ALL maxAttempts even after the +// task stops throwing (no early exit) — the unit test pins that behavior +// (it expects the neighbor count to keep growing past the threshold), so +// the port preserves it. TS's discoverEntryPoints option declares an +// optional excluded-nodes parameter, but the class never passes one (it +// filters locally) — the port takes a parameterless callback. +module; + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include // IWYU pragma: keep + +export module streamr.trackerlessnetwork.StreamPartNetworkSplitAvoidance; + +import streamr.dht.protos; + +import streamr.dht.Identifiers; +import streamr.logger.SLogger; +import streamr.trackerlessnetwork.DiscoveryLayerNode; +import streamr.utils.AbortController; +import streamr.utils.CoroutineHelper; + +// Hoisted from the former-header idiom (file scope, NOT exported). +using streamr::logger::SLogger; + +namespace { + +// TS exponentialRunOff: run `task` maxAttempts times with exponentially +// growing abortable waits between attempts; task failures are swallowed +// (the next attempt retries), and an abort ends the loop at the next +// checkpoint. +folly::coro::Task exponentialRunOff( + std::function()> task, + std::string description, + streamr::utils::AbortSignal& abortSignal, + std::chrono::milliseconds baseDelay = std::chrono::milliseconds{500}, + size_t maxAttempts = 6) { // NOLINT(readability-magic-numbers) + const auto cancellationToken = abortSignal.getCancellationToken(); + for (size_t i = 1; i <= maxAttempts; i++) { + if (abortSignal.aborted) { + co_return; + } + const auto factor = static_cast(1) << i; + const auto delay = baseDelay * factor; + try { + co_await task(); + } catch (const std::exception&) { + SLogger::trace( + description + " failed, retrying in " + + std::to_string(delay.count()) + " ms"); + } + try { + co_await streamr::utils::co_withCancellation( + cancellationToken, folly::coro::sleep(delay)); + } catch (const std::exception& err) { + SLogger::trace(err.what()); + } + } +} + +} // namespace + +export namespace streamr::trackerlessnetwork { + +using ::dht::PeerDescriptor; +using streamr::dht::DhtAddress; +using streamr::dht::Identifiers; +using streamr::trackerlessnetwork::discoverylayer::DiscoveryLayerNode; +using streamr::utils::AbortController; + +constexpr size_t minNeighborCount = 4; // TS MIN_NEIGHBOR_COUNT + +struct StreamPartNetworkSplitAvoidanceOptions { + DiscoveryLayerNode& discoveryLayerNode; + std::function>()> + discoverEntryPoints; + std::optional exponentialRunOfBaseDelay; +}; + +class StreamPartNetworkSplitAvoidance { +private: + static constexpr std::chrono::milliseconds defaultRunOffBaseDelay{500}; + + StreamPartNetworkSplitAvoidanceOptions options; + AbortController abortController; + std::set excludedNodes; + // Read by isRunning() from other threads while avoidNetworkSplit() is + // in flight on a pool thread. + std::atomic running = false; + +public: + explicit StreamPartNetworkSplitAvoidance( + StreamPartNetworkSplitAvoidanceOptions options) + : options(std::move(options)) {} + + folly::coro::Task avoidNetworkSplit() { + this->running = true; + co_await exponentialRunOff( + [this]() -> folly::coro::Task { + const auto discoveredEntryPoints = + co_await this->options.discoverEntryPoints(); + std::vector filteredEntryPoints; + for (const auto& peer : discoveredEntryPoints) { + if (!this->excludedNodes.contains( + Identifiers::getNodeIdFromPeerDescriptor(peer))) { + filteredEntryPoints.push_back(peer); + } + } + co_await this->options.discoveryLayerNode.joinDht( + filteredEntryPoints, + /*doRandomJoin=*/false, + /*retry=*/false); + if (this->options.discoveryLayerNode.getNeighborCount() < + minNeighborCount) { + // Exclude entry points that did not become neighbors + // (assumed offline). + const auto neighbors = + this->options.discoveryLayerNode.getNeighbors(); + for (const auto& peer : filteredEntryPoints) { + const auto isNeighbor = + [&peer](const PeerDescriptor& neighbor) { + return Identifiers::areEqualPeerDescriptors( + neighbor, peer); + }; + if (!std::ranges::any_of(neighbors, isNeighbor)) { + this->excludedNodes.insert( + Identifiers::getNodeIdFromPeerDescriptor(peer)); + } + } + throw std::runtime_error("Network split is still possible"); + } + }, + "avoid network split", + this->abortController.getSignal(), + this->options.exponentialRunOfBaseDelay.value_or( + defaultRunOffBaseDelay)); + this->running = false; + this->excludedNodes.clear(); + SLogger::trace("Network split avoided"); + } + + [[nodiscard]] bool isRunning() const { return this->running; } + + void destroy() { this->abortController.abort(); } +}; + +} // namespace streamr::trackerlessnetwork diff --git a/packages/streamr-trackerless-network/modules/StreamPartReconnect.cppm b/packages/streamr-trackerless-network/modules/StreamPartReconnect.cppm new file mode 100644 index 00000000..a274b303 --- /dev/null +++ b/packages/streamr-trackerless-network/modules/StreamPartReconnect.cppm @@ -0,0 +1,150 @@ +// Module streamr.trackerlessnetwork.StreamPartReconnect +// Ported from packages/trackerless-network/src/StreamPartReconnect.ts +// (v103.8.0-rc.3): after losing all neighbors, periodically re-fetches the +// stream's entry points from the DHT and rejoins the layer-1 network until +// a neighbor is found (the interval task then aborts itself). +// +// Port note: TS runs the loop via scheduleAtInterval and lets GC collect +// the detached closure; here the loop is owned by a scope that reconnect() +// re-arms and destroy() drains (awaited, never a blocking join on a pool +// thread) so no straggler iteration can outlive this object or the +// discovery layer node it references. +module; + +#include +#include +#include +#include + +#include // IWYU pragma: keep + +export module streamr.trackerlessnetwork.StreamPartReconnect; + +import streamr.dht.protos; + +import streamr.logger.SLogger; +import streamr.trackerlessnetwork.DiscoveryLayerNode; +import streamr.trackerlessnetwork.PeerDescriptorStoreManager; +import streamr.utils.AbortController; +import streamr.utils.CoroutineHelper; +import streamr.utils.SharedExecutors; + +// Hoisted from the former-header idiom (file scope, NOT exported). +using streamr::logger::SLogger; + +export namespace streamr::trackerlessnetwork { + +using streamr::trackerlessnetwork::controllayer::maxNodeCount; +using streamr::trackerlessnetwork::controllayer::PeerDescriptorStoreManager; +using streamr::trackerlessnetwork::discoverylayer::DiscoveryLayerNode; +using streamr::utils::AbortController; + +class StreamPartReconnect { +private: + static constexpr std::chrono::milliseconds defaultReconnectInterval{ + 30 * 1000}; + + DiscoveryLayerNode& discoveryLayerNode; + PeerDescriptorStoreManager& peerDescriptorStoreManager; + // Recreated by each reconnect() like TS; null until the first call. + std::unique_ptr abortController; + folly::coro::CancellableAsyncScope scope; + std::atomic scopeDrained = false; + + folly::coro::Task reconnectAttempt() { + const auto entryPoints = + co_await this->peerDescriptorStoreManager.fetchNodes(); + co_await this->discoveryLayerNode.joinDht(entryPoints); + // Is it necessary to store the node as an entry point here? (TS + // carries the same open question.) + if (!this->peerDescriptorStoreManager.isLocalNodeStored() && + entryPoints.size() < maxNodeCount) { + co_await this->peerDescriptorStoreManager.storeAndKeepLocalNode(); + } + if (this->discoveryLayerNode.getNeighborCount() > 0) { + this->abortController->abort(); + } + } + +public: + StreamPartReconnect( + DiscoveryLayerNode& discoveryLayerNode, + PeerDescriptorStoreManager& peerDescriptorStoreManager) + : discoveryLayerNode(discoveryLayerNode), + peerDescriptorStoreManager(peerDescriptorStoreManager) {} + + ~StreamPartReconnect() { this->drainScope(); } + StreamPartReconnect(const StreamPartReconnect&) = delete; + StreamPartReconnect& operator=(const StreamPartReconnect&) = delete; + StreamPartReconnect(StreamPartReconnect&&) = delete; + StreamPartReconnect& operator=(StreamPartReconnect&&) = delete; + + // TS scheduleAtInterval(task, timeout, true, signal): the first attempt + // is awaited by the caller, the recurring attempts run detached until a + // neighbor is found (the attempt aborts the controller) or destroy(). + folly::coro::Task reconnect( + std::chrono::milliseconds timeout = defaultReconnectInterval) { + if (this->scopeDrained) { + co_return; // destroyed; the joined scope cannot take new tasks + } + this->abortController = std::make_unique(); + const auto token = + this->abortController->getSignal().getCancellationToken(); + co_await this->reconnectAttempt(); + this->scope.add( + streamr::utils::co_withExecutor( + &streamr::utils::SharedExecutors::worker(), + folly::coro::co_invoke( + [this, token, timeout]() -> folly::coro::Task { + while (!token.isCancellationRequested()) { + try { + co_await streamr::utils::co_withCancellation( + token, folly::coro::sleep(timeout)); + } catch (const std::exception&) { + co_return; // aborted mid-sleep + } + if (token.isCancellationRequested()) { + co_return; + } + try { + co_await this->reconnectAttempt(); + } catch (const std::exception& err) { + SLogger::debug( + "reconnect attempt failed: " + + std::string(err.what())); + } + } + }))); + } + + [[nodiscard]] bool isRunning() const { + return this->abortController != nullptr && + !this->abortController->getSignal().aborted; + } + + void destroy() { + if (this->abortController != nullptr) { + this->abortController->abort(); + } + this->drainScope(); + } + +private: + // Blocking drain: must run on an owner thread, never a shared-pool + // worker (the repo-wide communicator-teardown contract). The loop's + // sleep is cancelled by the abort token, so abort() first keeps this + // prompt. + void drainScope() noexcept { + if (!this->scopeDrained.exchange(true)) { + if (this->abortController != nullptr) { + this->abortController->abort(); + } + try { + streamr::utils::blockingWait(this->scope.cancelAndJoinAsync()); + } catch (...) { // NOLINT(bugprone-empty-catch) must not throw + } + } + } +}; + +} // namespace streamr::trackerlessnetwork diff --git a/packages/streamr-trackerless-network/modules/control-layer/PeerDescriptorStoreManager.cppm b/packages/streamr-trackerless-network/modules/control-layer/PeerDescriptorStoreManager.cppm new file mode 100644 index 00000000..6761c6e1 --- /dev/null +++ b/packages/streamr-trackerless-network/modules/control-layer/PeerDescriptorStoreManager.cppm @@ -0,0 +1,227 @@ +// Module streamr.trackerlessnetwork.PeerDescriptorStoreManager +// Ported from packages/trackerless-network/src/control-layer/ +// PeerDescriptorStoreManager.ts (v103.8.0-rc.3): for each key there are +// usually 0..MAX_NODE_COUNT PeerDescriptors stored in the DHT; if there +// are fewer nodes, the local node's peer descriptor is stored (and then +// periodically re-stored) under the key. The DHT operations are taken as +// callbacks (TS options style) so the class stays independent of DhtNode +// and the unit test can substitute fakes. +// +// The public entry points are virtual (unlike TS, which force-casts an +// unrelated fake): StreamPartReconnect's test substitutes +// FakePeerDescriptorStoreManager by overriding them. +module; + +#include +#include +#include +#include +#include +#include +#include + +#include // IWYU pragma: keep + +#include + +export module streamr.trackerlessnetwork.PeerDescriptorStoreManager; + +import streamr.dht.protos; + +import streamr.dht.Identifiers; +import streamr.logger.SLogger; +import streamr.utils.AbortController; +import streamr.utils.CoroutineHelper; +import streamr.utils.SharedExecutors; + +// Hoisted from the former-header idiom (file scope, NOT exported). +using streamr::logger::SLogger; + +namespace { + +// TS parsePeerDescriptor: drop tombstoned entries, unpack the rest. +std::vector<::dht::PeerDescriptor> parsePeerDescriptors( + const std::vector<::dht::DataEntry>& dataEntries) { + std::vector<::dht::PeerDescriptor> result; + for (const auto& entry : dataEntries) { + if (entry.deleted()) { + continue; + } + ::dht::PeerDescriptor peerDescriptor; + if (entry.data().UnpackTo(&peerDescriptor)) { + result.push_back(std::move(peerDescriptor)); + } + } + return result; +} + +} // namespace + +export namespace streamr::trackerlessnetwork::controllayer { + +using ::dht::DataEntry; +using ::dht::PeerDescriptor; +using ::google::protobuf::Any; +using streamr::dht::DhtAddress; +using streamr::dht::Identifiers; +using streamr::utils::AbortController; + +constexpr size_t maxNodeCount = 8; // TS MAX_NODE_COUNT + +struct PeerDescriptorStoreManagerOptions { + DhtAddress key; + PeerDescriptor localPeerDescriptor; + std::optional storeInterval; + std::function>(DhtAddress)> + fetchDataFromDht; + std::function>( + DhtAddress, Any)> + storeDataToDht; + std::function(DhtAddress, bool)> deleteDataFromDht; +}; + +class PeerDescriptorStoreManager { +private: + static constexpr std::chrono::milliseconds defaultStoreInterval{60000}; + + PeerDescriptorStoreManagerOptions options; + AbortController abortController; + bool localNodeStored = false; + // Owns the keep-alive interval loop. TS detaches it (scheduleAtInterval) + // and lets GC collect the closure; in C++ a detached loop touching + // `this` is a use-after-free once the manager dies, so the loop lives + // in a scope drained by destroy()/the destructor (see the + // teardown-drain-ordering lessons: the drain is awaited, never a + // blocking join on a pool thread). + folly::coro::CancellableAsyncScope keepAliveScope; + std::atomic keepAliveScopeDrained = false; + + folly::coro::Task storeLocalNode() { + Any dataToStore; + dataToStore.PackFrom(this->options.localPeerDescriptor); + try { + co_await this->options.storeDataToDht( + this->options.key, std::move(dataToStore)); + } catch (const std::exception& err) { + SLogger::warn( + "Failed to store local node, key: " + this->options.key + " (" + + err.what() + ")"); + } + } + + folly::coro::Task keepLocalNodeAttempt() { + SLogger::trace( + "Attempting to keep local node, key: " + this->options.key); + try { + const auto discovered = co_await this->fetchNodes(); + const auto isLocal = [this](const PeerDescriptor& peerDescriptor) { + return Identifiers::areEqualPeerDescriptors( + peerDescriptor, this->options.localPeerDescriptor); + }; + if (discovered.size() < maxNodeCount || + std::ranges::any_of(discovered, isLocal)) { + co_await this->storeLocalNode(); + } + } catch (const std::exception& err) { + SLogger::debug( + "Failed to keep local node, key: " + this->options.key + " (" + + err.what() + ")"); + } + } + + // TS keepLocalNode = scheduleAtInterval(task, interval, false, signal): + // resolves immediately and the interval loop runs detached. Here the + // loop is a keepAliveScope task so teardown can drain it (see the + // member comment). + void keepLocalNode() { + const auto token = + this->abortController.getSignal().getCancellationToken(); + this->keepAliveScope.add( + streamr::utils::co_withExecutor( + &streamr::utils::SharedExecutors::worker(), + folly::coro::co_invoke( + [this, token]() -> folly::coro::Task { + const auto interval = + this->options.storeInterval.value_or( + defaultStoreInterval); + while (!token.isCancellationRequested()) { + try { + co_await streamr::utils::co_withCancellation( + token, folly::coro::sleep(interval)); + } catch (const std::exception&) { + co_return; // aborted mid-sleep + } + if (token.isCancellationRequested()) { + co_return; + } + co_await this->keepLocalNodeAttempt(); + } + }))); + } + +public: + explicit PeerDescriptorStoreManager( + PeerDescriptorStoreManagerOptions options) + : options(std::move(options)) {} + + virtual ~PeerDescriptorStoreManager() { this->drainKeepAliveScope(); } + PeerDescriptorStoreManager(const PeerDescriptorStoreManager&) = delete; + PeerDescriptorStoreManager& operator=(const PeerDescriptorStoreManager&) = + delete; + PeerDescriptorStoreManager(PeerDescriptorStoreManager&&) = delete; + PeerDescriptorStoreManager& operator=(PeerDescriptorStoreManager&&) = + delete; + + virtual folly::coro::Task> fetchNodes() { + SLogger::trace("Fetch data, key: " + this->options.key); + try { + const auto result = + co_await this->options.fetchDataFromDht(this->options.key); + co_return parsePeerDescriptors(result); + } catch (const std::exception&) { + co_return std::vector{}; + } + } + + virtual folly::coro::Task storeAndKeepLocalNode() { + if (this->abortController.getSignal().aborted) { + co_return; + } + this->localNodeStored = true; + co_await this->storeLocalNode(); + this->keepLocalNode(); + } + + [[nodiscard]] virtual bool isLocalNodeStored() const { + return this->localNodeStored; + } + + virtual folly::coro::Task destroy() { + this->abortController.abort(); + co_await this->options.deleteDataFromDht(this->options.key, false); + // Awaited (never a blocking join): safe on any thread, including + // pool workers. + if (!this->keepAliveScopeDrained.exchange(true)) { + co_await this->keepAliveScope.cancelAndJoinAsync(); + } + } + +protected: + // Backstop for owners that never call destroy(). Blocking: must run on + // an owner thread, never a shared-pool worker (the repo-wide + // communicator-teardown contract). Aborts first — the loop's sleep is + // cancelled by the abort token (which shadows the scope token), so a + // join without the abort would wait out the full store interval. + void drainKeepAliveScope() noexcept { + if (!this->keepAliveScopeDrained.exchange(true)) { + this->abortController.abort(); + try { + streamr::utils::blockingWait( + this->keepAliveScope.cancelAndJoinAsync()); + } catch (...) { // NOLINT(bugprone-empty-catch) must not throw + } + } + } +}; + +} // namespace streamr::trackerlessnetwork::controllayer diff --git a/packages/streamr-trackerless-network/modules/discovery-layer/DiscoveryLayerNode.cppm b/packages/streamr-trackerless-network/modules/discovery-layer/DiscoveryLayerNode.cppm new file mode 100644 index 00000000..c24271a1 --- /dev/null +++ b/packages/streamr-trackerless-network/modules/discovery-layer/DiscoveryLayerNode.cppm @@ -0,0 +1,78 @@ +// Module streamr.trackerlessnetwork.DiscoveryLayerNode +// Ported from packages/trackerless-network/src/discovery-layer/ +// DiscoveryLayerNode.ts (v103.8.0-rc.3): the control layer's view of a +// stream part's layer-1 DHT node. The production implementation wraps a +// streamr::dht::DhtNode (phase C6's createDiscoveryLayerNode); unit tests +// substitute streamr.trackerlessnetwork.MockDiscoveryLayerNode. +module; + +#include +#include +#include +#include + +#include // IWYU pragma: keep + +export module streamr.trackerlessnetwork.DiscoveryLayerNode; + +import streamr.dht.protos; + +// export import: the interface returns ClosestRingPeerDescriptors, so +// every consumer needs the declaration visible. +export import streamr.dht.DhtNodeRpcRemote; +import streamr.dht.Identifiers; +import streamr.eventemitter.EventEmitter; +import streamr.utils.CoroutineHelper; + +export namespace streamr::trackerlessnetwork::discoverylayer { + +using ::dht::PeerDescriptor; +// TS RingContacts ({ left, right } of PeerDescriptors) — the dht package's +// ClosestRingPeerDescriptors is that exact shape and what +// DhtNode::getRingContacts() returns. +using streamr::dht::ClosestRingPeerDescriptors; +using streamr::dht::DhtAddress; +using streamr::eventemitter::Event; +using streamr::eventemitter::EventEmitter; + +namespace discoverylayernodeevents { + +struct ManualRejoinRequired : Event<> {}; +struct NearbyContactAdded : Event {}; +struct NearbyContactRemoved : Event {}; +struct RandomContactAdded : Event {}; +struct RandomContactRemoved : Event {}; +struct RingContactAdded : Event {}; +struct RingContactRemoved : Event {}; + +} // namespace discoverylayernodeevents + +using DiscoveryLayerNodeEvents = std::tuple< + discoverylayernodeevents::ManualRejoinRequired, + discoverylayernodeevents::NearbyContactAdded, + discoverylayernodeevents::NearbyContactRemoved, + discoverylayernodeevents::RandomContactAdded, + discoverylayernodeevents::RandomContactRemoved, + discoverylayernodeevents::RingContactAdded, + discoverylayernodeevents::RingContactRemoved>; + +class DiscoveryLayerNode : public EventEmitter { +public: + virtual void removeContact(const DhtAddress& nodeId) = 0; + [[nodiscard]] virtual std::vector getClosestContacts( + std::optional maxCount = std::nullopt) = 0; + [[nodiscard]] virtual std::vector getRandomContacts( + std::optional maxCount = std::nullopt) = 0; + [[nodiscard]] virtual ClosestRingPeerDescriptors getRingContacts() = 0; + [[nodiscard]] virtual std::vector getNeighbors() = 0; + [[nodiscard]] virtual size_t getNeighborCount() = 0; + virtual folly::coro::Task joinDht( + std::vector entryPoints, + bool doRandomJoin = true, + bool retry = true) = 0; + virtual folly::coro::Task joinRing() = 0; + virtual folly::coro::Task start() = 0; + virtual folly::coro::Task stop() = 0; +}; + +} // namespace streamr::trackerlessnetwork::discoverylayer diff --git a/packages/streamr-trackerless-network/test/unit/PeerDescriptorStoreManagerTest.cpp b/packages/streamr-trackerless-network/test/unit/PeerDescriptorStoreManagerTest.cpp new file mode 100644 index 00000000..8f5b7827 --- /dev/null +++ b/packages/streamr-trackerless-network/test/unit/PeerDescriptorStoreManagerTest.cpp @@ -0,0 +1,122 @@ +// Ported from packages/trackerless-network/test/unit/ +// PeerDescriptorStoreManager.test.ts (v103.8.0-rc.3). The DHT operations +// are fakes counting store calls; the last case is timing-based like the +// TS original (storeInterval 2 s, wait 4.5 s → two extra keep-alive +// stores). +#include +#include +#include +#include +#include +#include +#include "packages/dht/protos/DhtRpc.pb.h" + +#include // IWYU pragma: keep + +import streamr.dht.Identifiers; +import streamr.dht.protos; +import streamr.trackerlessnetwork.PeerDescriptorStoreManager; +import streamr.trackerlessnetwork.TestUtils; +import streamr.utils.CoroutineHelper; + +using ::dht::DataEntry; +using ::dht::PeerDescriptor; +using streamr::dht::Identifiers; +using streamr::trackerlessnetwork::controllayer::PeerDescriptorStoreManager; +using streamr::trackerlessnetwork::controllayer:: + PeerDescriptorStoreManagerOptions; +using streamr::trackerlessnetwork::testutils::createMockPeerDescriptor; +using streamr::utils::blockingWait; + +namespace { +constexpr std::chrono::milliseconds testStoreInterval{2000}; +} // namespace + +class PeerDescriptorStoreManagerTest : public ::testing::Test { +protected: + PeerDescriptor peerDescriptor = createMockPeerDescriptor(); + PeerDescriptor deletedPeerDescriptor = createMockPeerDescriptor(); + std::atomic storeCalled = 0; + std::unique_ptr withData; + std::unique_ptr withoutData; + + [[nodiscard]] DataEntry createFakeData( + const PeerDescriptor& descriptor, bool deleted) const { + DataEntry entry; + entry.set_key("\x01\x02\x03"); + entry.mutable_data()->PackFrom(descriptor); + entry.set_creator(descriptor.nodeid()); + entry.set_ttl(1000); // NOLINT(readability-magic-numbers) + entry.set_stale(false); + entry.set_deleted(deleted); + return entry; + } + + void SetUp() override { + const auto key = Identifiers::createRandomDhtAddress(); + auto storeDataToDht = [this](auto /*key*/, auto /*data*/) + -> folly::coro::Task> { + this->storeCalled++; + co_return std::vector{this->peerDescriptor}; + }; + auto deleteDataFromDht = [](auto /*key*/, bool /*waitForCompletion*/) + -> folly::coro::Task { co_return; }; + this->withData = std::make_unique( + PeerDescriptorStoreManagerOptions{ + .key = key, + .localPeerDescriptor = this->peerDescriptor, + .storeInterval = testStoreInterval, + .fetchDataFromDht = [this](auto /*key*/) + -> folly::coro::Task> { + co_return std::vector{ + this->createFakeData(this->peerDescriptor, false), + this->createFakeData( + this->deletedPeerDescriptor, true)}; + }, + .storeDataToDht = storeDataToDht, + .deleteDataFromDht = deleteDataFromDht}); + this->withoutData = std::make_unique( + PeerDescriptorStoreManagerOptions{ + .key = key, + .localPeerDescriptor = this->peerDescriptor, + .storeInterval = testStoreInterval, + .fetchDataFromDht = [](auto /*key*/) + -> folly::coro::Task> { + co_return std::vector{}; + }, + .storeDataToDht = storeDataToDht, + .deleteDataFromDht = deleteDataFromDht}); + } + + void TearDown() override { blockingWait(this->withData->destroy()); } +}; + +TEST_F(PeerDescriptorStoreManagerTest, FetchNodesFiltersDeletedData) { + const auto result = blockingWait(this->withData->fetchNodes()); + ASSERT_EQ(result.size(), 1); + EXPECT_EQ( + Identifiers::areEqualPeerDescriptors(result[0], this->peerDescriptor), + true); +} + +TEST_F(PeerDescriptorStoreManagerTest, FetchNodesWithoutResults) { + const auto result = blockingWait(this->withoutData->fetchNodes()); + EXPECT_EQ(result.size(), 0); +} + +TEST_F(PeerDescriptorStoreManagerTest, StoreOnStreamWithoutSaturatedCount) { + blockingWait(this->withData->storeAndKeepLocalNode()); + EXPECT_EQ(this->storeCalled, 1); + EXPECT_EQ(this->withData->isLocalNodeStored(), true); +} + +TEST_F(PeerDescriptorStoreManagerTest, WillKeepStoredUntilDestroyed) { + blockingWait(this->withData->storeAndKeepLocalNode()); + EXPECT_EQ(this->storeCalled, 1); + EXPECT_EQ(this->withData->isLocalNodeStored(), true); + // storeInterval is 2 s: after 4.5 s the keep-alive loop has stored two + // more times (TS waits the same 4.5 s). + std::this_thread::sleep_for(std::chrono::milliseconds(4500)); + blockingWait(this->withData->destroy()); + EXPECT_EQ(this->storeCalled, 3); +} diff --git a/packages/streamr-trackerless-network/test/unit/StreamPartNetworkSplitAvoidanceTest.cpp b/packages/streamr-trackerless-network/test/unit/StreamPartNetworkSplitAvoidanceTest.cpp new file mode 100644 index 00000000..53be37a4 --- /dev/null +++ b/packages/streamr-trackerless-network/test/unit/StreamPartNetworkSplitAvoidanceTest.cpp @@ -0,0 +1,54 @@ +// Ported from packages/trackerless-network/test/unit/ +// StreamPartNetworkSplitAvoidance.test.ts (v103.8.0-rc.3): each +// discoverEntryPoints call grows the mock's k-bucket by one random peer; +// the run-off runs all its attempts, so the final neighbor count exceeds +// (not merely reaches) minNeighborCount. +#include +#include +#include +#include +#include "packages/dht/protos/DhtRpc.pb.h" + +#include // IWYU pragma: keep + +import streamr.dht.protos; +import streamr.trackerlessnetwork.MockDiscoveryLayerNode; +import streamr.trackerlessnetwork.StreamPartNetworkSplitAvoidance; +import streamr.utils.CoroutineHelper; + +using ::dht::PeerDescriptor; +using streamr::trackerlessnetwork::minNeighborCount; +using streamr::trackerlessnetwork::StreamPartNetworkSplitAvoidance; +using streamr::trackerlessnetwork::StreamPartNetworkSplitAvoidanceOptions; +using streamr::trackerlessnetwork::testutils::MockDiscoveryLayerNode; +using streamr::utils::blockingWait; + +class StreamPartNetworkSplitAvoidanceTest : public ::testing::Test { +protected: + MockDiscoveryLayerNode discoveryLayerNode; + std::unique_ptr avoidance; + + void SetUp() override { + this->avoidance = std::make_unique( + StreamPartNetworkSplitAvoidanceOptions{ + .discoveryLayerNode = this->discoveryLayerNode, + .discoverEntryPoints = + [this]() -> folly::coro::Task> { + this->discoveryLayerNode.addNewRandomPeerToKBucket(); + co_return this->discoveryLayerNode.getNeighbors(); + }, + .exponentialRunOfBaseDelay = std::chrono::milliseconds(1)}); + } + + void TearDown() override { + blockingWait(this->discoveryLayerNode.stop()); + this->avoidance->destroy(); + } +}; + +TEST_F( + StreamPartNetworkSplitAvoidanceTest, + RunsAvoidanceUntilNumberOfNeighborsIsAboveMinNeighborCount) { + blockingWait(this->avoidance->avoidNetworkSplit()); + EXPECT_GT(this->discoveryLayerNode.getNeighborCount(), minNeighborCount); +} diff --git a/packages/streamr-trackerless-network/test/unit/StreamPartReconnectTest.cpp b/packages/streamr-trackerless-network/test/unit/StreamPartReconnectTest.cpp new file mode 100644 index 00000000..2b9de77d --- /dev/null +++ b/packages/streamr-trackerless-network/test/unit/StreamPartReconnectTest.cpp @@ -0,0 +1,101 @@ +// Ported from packages/trackerless-network/test/unit/ +// StreamPartReconnect.test.ts (v103.8.0-rc.3). The fake store manager is +// ported inline from test/utils/fake/FakePeerDescriptorStoreManager.ts +// (its only user); the TS fake force-casts an unrelated class, here it +// overrides the manager's virtual entry points. +#include +#include +#include +#include +#include "packages/dht/protos/DhtRpc.pb.h" + +#include // IWYU pragma: keep + +import streamr.dht.Identifiers; +import streamr.dht.protos; +import streamr.trackerlessnetwork.MockDiscoveryLayerNode; +import streamr.trackerlessnetwork.PeerDescriptorStoreManager; +import streamr.trackerlessnetwork.StreamPartReconnect; +import streamr.trackerlessnetwork.TestUtils; +import streamr.utils.CoroutineHelper; +import streamr.utils.waitForCondition; + +using ::dht::DataEntry; +using ::dht::PeerDescriptor; +using streamr::dht::Identifiers; +using streamr::trackerlessnetwork::StreamPartReconnect; +using streamr::trackerlessnetwork::controllayer::PeerDescriptorStoreManager; +using streamr::trackerlessnetwork::controllayer:: + PeerDescriptorStoreManagerOptions; +using streamr::trackerlessnetwork::testutils::createMockPeerDescriptor; +using streamr::trackerlessnetwork::testutils::MockDiscoveryLayerNode; +using streamr::utils::blockingWait; +using streamr::utils::waitForCondition; + +namespace { + +constexpr auto untilTimeout = std::chrono::seconds(15); +constexpr auto untilPollInterval = std::chrono::milliseconds(50); + +class FakePeerDescriptorStoreManager : public PeerDescriptorStoreManager { +private: + std::vector nodes; + +public: + FakePeerDescriptorStoreManager() + : PeerDescriptorStoreManager( + PeerDescriptorStoreManagerOptions{ + .key = Identifiers::createRandomDhtAddress(), + .localPeerDescriptor = createMockPeerDescriptor(), + .storeInterval = std::nullopt, + .fetchDataFromDht = [](auto /*key*/) + -> folly::coro::Task> { + co_return std::vector{}; + }, + .storeDataToDht = [](auto /*key*/, auto /*data*/) + -> folly::coro::Task> { + co_return std::vector{}; + }, + .deleteDataFromDht = [](auto /*key*/, + bool /*waitForCompletion*/) + -> folly::coro::Task { co_return; }}) {} + + void setNodes(std::vector newNodes) { + this->nodes = std::move(newNodes); + } + + folly::coro::Task> fetchNodes() override { + co_return this->nodes; + } + + folly::coro::Task storeAndKeepLocalNode() override { co_return; } + + [[nodiscard]] bool isLocalNodeStored() const override { return true; } +}; + +} // namespace + +class StreamPartReconnectTest : public ::testing::Test { +protected: + FakePeerDescriptorStoreManager peerDescriptorStoreManager; + MockDiscoveryLayerNode discoveryLayerNode; + std::unique_ptr streamPartReconnect; + + void SetUp() override { + this->streamPartReconnect = std::make_unique( + this->discoveryLayerNode, this->peerDescriptorStoreManager); + } + + void TearDown() override { this->streamPartReconnect->destroy(); } +}; + +TEST_F(StreamPartReconnectTest, HappyPath) { + blockingWait( + this->streamPartReconnect->reconnect(std::chrono::milliseconds(1000))); + EXPECT_EQ(this->streamPartReconnect->isRunning(), true); + this->discoveryLayerNode.addNewRandomPeerToKBucket(); + EXPECT_NO_THROW(blockingWait(waitForCondition( + [this]() { return !this->streamPartReconnect->isRunning(); }, + untilTimeout, + untilPollInterval))); +} diff --git a/packages/streamr-trackerless-network/test/utils/MockDiscoveryLayerNode.cppm b/packages/streamr-trackerless-network/test/utils/MockDiscoveryLayerNode.cppm new file mode 100644 index 00000000..8f919dcb --- /dev/null +++ b/packages/streamr-trackerless-network/test/utils/MockDiscoveryLayerNode.cppm @@ -0,0 +1,97 @@ +// Module streamr.trackerlessnetwork.MockDiscoveryLayerNode +// Ported from packages/trackerless-network/test/utils/mock/ +// MockDiscoveryLayerNode.ts (v103.8.0-rc.3): an in-memory +// DiscoveryLayerNode whose k-bucket grows only via +// addNewRandomPeerToKBucket(); join/start/stop are no-ops. +module; + +#include +#include +#include +#include +#include + +#include // IWYU pragma: keep + +export module streamr.trackerlessnetwork.MockDiscoveryLayerNode; + +import streamr.dht.protos; + +import streamr.dht.Identifiers; +import streamr.trackerlessnetwork.DiscoveryLayerNode; +import streamr.trackerlessnetwork.TestUtils; +import streamr.utils.CoroutineHelper; + +export namespace streamr::trackerlessnetwork::testutils { + +using ::dht::PeerDescriptor; +using streamr::dht::ClosestRingPeerDescriptors; +using streamr::dht::DhtAddress; +using streamr::trackerlessnetwork::discoverylayer::DiscoveryLayerNode; + +class MockDiscoveryLayerNode : public DiscoveryLayerNode { +private: + // TS runs single-threaded; here the test thread grows the k-bucket + // while a detached loop (e.g. StreamPartReconnect's) reads it on a + // pool thread. + std::mutex mutex; + std::vector kbucketPeers; + std::vector closestContacts; + std::vector randomContacts; + +public: + void removeContact(const DhtAddress& /*nodeId*/) override {} + + [[nodiscard]] std::vector getClosestContacts( + std::optional /*maxCount*/) override { + return this->closestContacts; + } + + void setClosestContacts(std::vector contacts) { + this->closestContacts = std::move(contacts); + } + + [[nodiscard]] std::vector getRandomContacts( + std::optional /*maxCount*/) override { + return this->randomContacts; + } + + void setRandomContacts(std::vector contacts) { + this->randomContacts = std::move(contacts); + } + + [[nodiscard]] ClosestRingPeerDescriptors getRingContacts() override { + return ClosestRingPeerDescriptors{.left = {}, .right = {}}; + } + + [[nodiscard]] std::vector getNeighbors() override { + std::scoped_lock lock(this->mutex); + return this->kbucketPeers; + } + + [[nodiscard]] size_t getNeighborCount() override { + std::scoped_lock lock(this->mutex); + return this->kbucketPeers.size(); + } + + void addNewRandomPeerToKBucket() { + std::scoped_lock lock(this->mutex); + this->kbucketPeers.push_back( + streamr::trackerlessnetwork::testutils::createMockPeerDescriptor()); + } + + folly::coro::Task joinDht( + std::vector /*entryPoints*/, + bool /*doRandomJoin*/, + bool /*retry*/) override { + co_return; + } + + folly::coro::Task joinRing() override { co_return; } + + folly::coro::Task start() override { co_return; } + + folly::coro::Task stop() override { co_return; } +}; + +} // namespace streamr::trackerlessnetwork::testutils diff --git a/trackerless-network-completion-plan.md b/trackerless-network-completion-plan.md index b7e58c22..df922ea8 100644 --- a/trackerless-network-completion-plan.md +++ b/trackerless-network-completion-plan.md @@ -565,6 +565,19 @@ New classes: `PeerDescriptorStoreManager` (store/fetch stream entry points under `integration/streamEntryPointReplacing.test.ts`, `integration/joining-streams-on-offline-peers.test.ts`. +*Implemented (phase-C4 PR):* the three classes plus the +`DiscoveryLayerNode` interface (`modules/discovery-layer/`, TS RingContacts += the dht package's `ClosestRingPeerDescriptors`) and a +`MockDiscoveryLayerNode` test util; `unit/StreamPartReconnect.test.ts` +(exists in TS though unlisted above) ported as well. Deviations: the TS +classes detach their interval loops (`scheduleAtInterval`) and rely on GC — +here the loops are owned by a per-instance `CancellableAsyncScope` drained +in `destroy()` (awaited, so no pool-thread parking) with a blocking +destructor backstop, per the teardown-drain-ordering lessons; +`discoverEntryPoints` is a parameterless callback (TS declares an optional +excluded-nodes parameter it never passes). The three integration tests +require `NetworkNode`/`createNetworkNode` and move to milestone C6/C8. + **Phase C5 — ContentDeliveryManager and proxy server side.** New/extended: `ContentDeliveryManager` (join/leave/broadcast per stream part, discovery-layer node creation, proxy setup, `suppressOwnMessageLoopback`), extend `ProxyConnectionRpcLocal` and From bba12dcb65258a23f25c8cb7375f8066622b1248 Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Sun, 12 Jul 2026 01:46:04 +0300 Subject: [PATCH 2/3] Phase C4 lint fixes: consume dht protos via BMI only in the new tests, named test constants clangd 22 flags every member call on the generated dht types as ambiguous when a test TU both textually includes DhtRpc.pb.h and imports streamr.dht.protos (documented duplicate-decl merge failure) - drop the textual include. Also: static createFakeData, named constants for the keep-alive observation window, fake TTL and test reconnect interval. Co-Authored-By: Claude Fable 5 --- .../test/unit/PeerDescriptorStoreManagerTest.cpp | 15 ++++++++++----- .../unit/StreamPartNetworkSplitAvoidanceTest.cpp | 4 +++- .../test/unit/StreamPartReconnectTest.cpp | 8 +++++--- 3 files changed, 18 insertions(+), 9 deletions(-) diff --git a/packages/streamr-trackerless-network/test/unit/PeerDescriptorStoreManagerTest.cpp b/packages/streamr-trackerless-network/test/unit/PeerDescriptorStoreManagerTest.cpp index 8f5b7827..5afb8e77 100644 --- a/packages/streamr-trackerless-network/test/unit/PeerDescriptorStoreManagerTest.cpp +++ b/packages/streamr-trackerless-network/test/unit/PeerDescriptorStoreManagerTest.cpp @@ -3,13 +3,15 @@ // are fakes counting store calls; the last case is timing-based like the // TS original (storeInterval 2 s, wait 4.5 s → two extra keep-alive // stores). +// The generated dht protos come ONLY from `import streamr.dht.protos` — a +// textual DhtRpc.pb.h include alongside the BMI makes clangd flag every +// member call on those types as ambiguous. #include #include #include #include #include #include -#include "packages/dht/protos/DhtRpc.pb.h" #include // IWYU pragma: keep @@ -30,6 +32,9 @@ using streamr::utils::blockingWait; namespace { constexpr std::chrono::milliseconds testStoreInterval{2000}; +// TS waits 4.5 s: two keep-alive ticks of the 2 s interval fit. +constexpr std::chrono::milliseconds keepAliveObservationTime{4500}; +constexpr uint32_t fakeDataTtlMs = 1000; } // namespace class PeerDescriptorStoreManagerTest : public ::testing::Test { @@ -40,13 +45,13 @@ class PeerDescriptorStoreManagerTest : public ::testing::Test { std::unique_ptr withData; std::unique_ptr withoutData; - [[nodiscard]] DataEntry createFakeData( - const PeerDescriptor& descriptor, bool deleted) const { + [[nodiscard]] static DataEntry createFakeData( + const PeerDescriptor& descriptor, bool deleted) { DataEntry entry; entry.set_key("\x01\x02\x03"); entry.mutable_data()->PackFrom(descriptor); entry.set_creator(descriptor.nodeid()); - entry.set_ttl(1000); // NOLINT(readability-magic-numbers) + entry.set_ttl(fakeDataTtlMs); entry.set_stale(false); entry.set_deleted(deleted); return entry; @@ -116,7 +121,7 @@ TEST_F(PeerDescriptorStoreManagerTest, WillKeepStoredUntilDestroyed) { EXPECT_EQ(this->withData->isLocalNodeStored(), true); // storeInterval is 2 s: after 4.5 s the keep-alive loop has stored two // more times (TS waits the same 4.5 s). - std::this_thread::sleep_for(std::chrono::milliseconds(4500)); + std::this_thread::sleep_for(keepAliveObservationTime); blockingWait(this->withData->destroy()); EXPECT_EQ(this->storeCalled, 3); } diff --git a/packages/streamr-trackerless-network/test/unit/StreamPartNetworkSplitAvoidanceTest.cpp b/packages/streamr-trackerless-network/test/unit/StreamPartNetworkSplitAvoidanceTest.cpp index 53be37a4..cd004dc4 100644 --- a/packages/streamr-trackerless-network/test/unit/StreamPartNetworkSplitAvoidanceTest.cpp +++ b/packages/streamr-trackerless-network/test/unit/StreamPartNetworkSplitAvoidanceTest.cpp @@ -3,11 +3,13 @@ // discoverEntryPoints call grows the mock's k-bucket by one random peer; // the run-off runs all its attempts, so the final neighbor count exceeds // (not merely reaches) minNeighborCount. +// The generated dht protos come ONLY from `import streamr.dht.protos` — a +// textual DhtRpc.pb.h include alongside the BMI makes clangd flag every +// member call on those types as ambiguous. #include #include #include #include -#include "packages/dht/protos/DhtRpc.pb.h" #include // IWYU pragma: keep diff --git a/packages/streamr-trackerless-network/test/unit/StreamPartReconnectTest.cpp b/packages/streamr-trackerless-network/test/unit/StreamPartReconnectTest.cpp index 2b9de77d..98f7e274 100644 --- a/packages/streamr-trackerless-network/test/unit/StreamPartReconnectTest.cpp +++ b/packages/streamr-trackerless-network/test/unit/StreamPartReconnectTest.cpp @@ -3,11 +3,13 @@ // ported inline from test/utils/fake/FakePeerDescriptorStoreManager.ts // (its only user); the TS fake force-casts an unrelated class, here it // overrides the manager's virtual entry points. +// The generated dht protos come ONLY from `import streamr.dht.protos` — a +// textual DhtRpc.pb.h include alongside the BMI makes clangd flag every +// member call on those types as ambiguous. #include #include #include #include -#include "packages/dht/protos/DhtRpc.pb.h" #include // IWYU pragma: keep @@ -36,6 +38,7 @@ namespace { constexpr auto untilTimeout = std::chrono::seconds(15); constexpr auto untilPollInterval = std::chrono::milliseconds(50); +constexpr auto testReconnectInterval = std::chrono::milliseconds(1000); class FakePeerDescriptorStoreManager : public PeerDescriptorStoreManager { private: @@ -90,8 +93,7 @@ class StreamPartReconnectTest : public ::testing::Test { }; TEST_F(StreamPartReconnectTest, HappyPath) { - blockingWait( - this->streamPartReconnect->reconnect(std::chrono::milliseconds(1000))); + blockingWait(this->streamPartReconnect->reconnect(testReconnectInterval)); EXPECT_EQ(this->streamPartReconnect->isRunning(), true); this->discoveryLayerNode.addNewRandomPeerToKBucket(); EXPECT_NO_THROW(blockingWait(waitForCondition( From b1dfe5feeb74d7f24bda1ac016a7dee0bcc163f7 Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Sun, 12 Jul 2026 02:05:50 +0300 Subject: [PATCH 3/3] Phase C4 lint fix: call static createFakeData without this-> Co-Authored-By: Claude Fable 5 --- .../test/unit/PeerDescriptorStoreManagerTest.cpp | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/packages/streamr-trackerless-network/test/unit/PeerDescriptorStoreManagerTest.cpp b/packages/streamr-trackerless-network/test/unit/PeerDescriptorStoreManagerTest.cpp index 5afb8e77..1846a919 100644 --- a/packages/streamr-trackerless-network/test/unit/PeerDescriptorStoreManagerTest.cpp +++ b/packages/streamr-trackerless-network/test/unit/PeerDescriptorStoreManagerTest.cpp @@ -74,9 +74,8 @@ class PeerDescriptorStoreManagerTest : public ::testing::Test { .fetchDataFromDht = [this](auto /*key*/) -> folly::coro::Task> { co_return std::vector{ - this->createFakeData(this->peerDescriptor, false), - this->createFakeData( - this->deletedPeerDescriptor, true)}; + createFakeData(this->peerDescriptor, false), + createFakeData(this->deletedPeerDescriptor, true)}; }, .storeDataToDht = storeDataToDht, .deleteDataFromDht = deleteDataFromDht});