Phase C4: entry points in the DHT (PeerDescriptorStoreManager, StreamPartNetworkSplitAvoidance, StreamPartReconnect)

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 $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/src/proto>)
     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

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 <algorithm>
+#include <atomic>
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <exception>
+#include <functional>
+#include <optional>
+#include <set>
+#include <stdexcept>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include <coroutine> // 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<void> exponentialRunOff(
+    std::function<folly::coro::Task<void>()> 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<int64_t>(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<folly::coro::Task<std::vector<PeerDescriptor>>()>
+        discoverEntryPoints;
+    std::optional<std::chrono::milliseconds> exponentialRunOfBaseDelay;
+};
+
+class StreamPartNetworkSplitAvoidance {
+private:
+    static constexpr std::chrono::milliseconds defaultRunOffBaseDelay{500};
+
+    StreamPartNetworkSplitAvoidanceOptions options;
+    AbortController abortController;
+    std::set<DhtAddress> excludedNodes;
+    // Read by isRunning() from other threads while avoidNetworkSplit() is
+    // in flight on a pool thread.
+    std::atomic<bool> running = false;
+
+public:
+    explicit StreamPartNetworkSplitAvoidance(
+        StreamPartNetworkSplitAvoidanceOptions options)
+        : options(std::move(options)) {}
+
+    folly::coro::Task<void> avoidNetworkSplit() {
+        this->running = true;
+        co_await exponentialRunOff(
+            [this]() -> folly::coro::Task<void> {
+                const auto discoveredEntryPoints =
+                    co_await this->options.discoverEntryPoints();
+                std::vector<PeerDescriptor> 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

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 <atomic>
+#include <chrono>
+#include <exception>
+#include <memory>
+
+#include <coroutine> // 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> abortController;
+    folly::coro::CancellableAsyncScope scope;
+    std::atomic<bool> scopeDrained = false;
+
+    folly::coro::Task<void> 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<void> reconnect(
+        std::chrono::milliseconds timeout = defaultReconnectInterval) {
+        if (this->scopeDrained) {
+            co_return; // destroyed; the joined scope cannot take new tasks
+        }
+        this->abortController = std::make_unique<AbortController>();
+        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<void> {
+                        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