Phase A7: discovery and join (PeerDiscovery, DiscoverySession, RingDiscoverySession) (#65)

Ports the DHT discovery/join layer from streamr/network af966cf03
(v103.8.0-rc.3).

- DiscoverySession: one Kademlia lookup toward a target id. The TS
  Gate/withTimeout/.finally-recursion fan-out becomes `parallelism` worker
  coroutines pulling from the shared frontier under the session mutex,
  awaited under one folly::coro::timeout (a `done` flag stands in for the
  Gate). Discovery is I/O-bound, so it is not offloaded to a dedicated
  executor like the Router's routing-table work.
- RingDiscoverySession: the ring-topology counterpart.
- PeerDiscovery: join orchestration (per entry point a discovery session
  toward the local id plus an optional distant/bit-flipped session, retry
  when no neighbours are found, periodic recovery once joined). shared_ptr-
  owned; retry/recovery are detached onto a dedicated worker executor (2nd-
  class maintenance off the caller/delivery threads) and torn down by the
  join abort signal.

Supporting changes:
- streamr.utils.scheduleAtInterval (new); ExecutorHelper now exports
  folly::Executor.
- DhtNodeRpcRemote::getClosestPeers/getClosestRingPeers made virtual (like
  ping) so the discovery test can substitute them.
- PeerManager::getNearbyContacts(); streamr.dht.consts
  (CONTROL_LAYER_NODE_SERVICE_ID).

Tests: ports unit/DiscoverySession.test.ts with the createTestTopology
partition-merge helper. The two join integration tests
(DhtJoinPeerDiscovery, MultipleEntryPointJoining) build full DhtNodes over
the Simulator and are deferred to phase A8 (DhtNode assembly).

170/170 dht unit, 12/12 dht integration, 51/51 utils unit; lint green.

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: ptesavol

packages/streamr-dht/CMakeLists.txt

@@ -220,6 +220,7 @@ if(NOT IOS AND STREAMR_MODULES_SUPPORTED)
         test/unit/LocalDataStoreTest.cpp
         test/unit/StoreRpcLocalTest.cpp
         test/unit/StoreManagerTest.cpp
+        test/unit/DiscoverySessionTest.cpp
     )
 
     target_include_directories(streamr-dht-test-unit

packages/streamr-dht/modules/dht/DhtNodeRpcRemote.cppm

@@ -95,7 +95,8 @@ public:
             this->getPeerDescriptor());
     }
 
-    folly::coro::Task<std::vector<PeerDescriptor>> getClosestPeers(
+    // Virtual (like ping()) so tests can substitute a deterministic result.
+    virtual folly::coro::Task<std::vector<PeerDescriptor>> getClosestPeers(
         const DhtAddress& nodeId) {
         ClosestPeersRequest request;
         request.set_nodeid(Identifiers::getRawFromDhtAddress(nodeId));
@@ -112,7 +113,8 @@ public:
         co_return peers;
     }
 
-    folly::coro::Task<ClosestRingPeerDescriptors> getClosestRingPeers(
+    // Virtual (like ping()) so tests can substitute a deterministic result.
+    virtual folly::coro::Task<ClosestRingPeerDescriptors> getClosestRingPeers(
         const RingIdRaw& ringIdRaw) {
         ClosestRingPeersRequest request;
         request.set_ringid(ringIdRaw);

packages/streamr-dht/modules/dht/PeerManager.cppm

@@ -429,6 +429,21 @@ public:
         return this->nearbyContacts->getSize(excludedNodeIds);
     }
 
+    // TS returns the SortedContactList itself; the only consumer
+    // (DiscoverySession) maps it straight to peer descriptors, so this returns
+    // the descriptors directly and keeps the list access under the mutex.
+    [[nodiscard]] std::vector<PeerDescriptor> getNearbyContacts() {
+        std::scoped_lock lock(this->mutex);
+        const auto contacts =
+            this->nearbyContacts->getAllContactsInUndefinedOrder();
+        std::vector<PeerDescriptor> result;
+        result.reserve(contacts.size());
+        for (const auto& contact : contacts) {
+            result.push_back(contact->getPeerDescriptor());
+        }
+        return result;
+    }
+
     [[nodiscard]] size_t getNeighborCount() {
         std::scoped_lock lock(this->mutex);
         return this->neighbors->count();

packages/streamr-dht/modules/dht/consts.cppm

@@ -0,0 +1,17 @@
+// Module streamr.dht.consts
+// Ported from packages/dht/src/dht/consts.ts (v103.8.0-rc.3).
+module;
+
+#include <string>
+
+export module streamr.dht.consts;
+
+import streamr.dht.Identifiers;
+
+export namespace streamr::dht {
+
+// TODO (from TS): move this to the trackerless-network package and make the
+// serviceId a required parameter.
+inline const ServiceID CONTROL_LAYER_NODE_SERVICE_ID{"layer0"};
+
+} // namespace streamr::dht

packages/streamr-dht/modules/dht/discovery/DiscoverySession.cppm

@@ -0,0 +1,262 @@
+// Module streamr.dht.DiscoverySession
+// Ported from packages/dht/src/dht/discovery/DiscoverySession.ts
+// (v103.8.0-rc.3). One Kademlia-style lookup toward a target id: it keeps
+// up to `parallelism` getClosestPeers requests in flight against the closest
+// uncontacted nearby contacts, feeds the returned peers back into the
+// PeerManager, and finishes once the frontier is exhausted or `noProgressLimit`
+// consecutive requests fail to bring a closer neighbour.
+//
+// TS drives the fan-out through promise continuations: each request's
+// `.finally` re-invokes findMoreContacts, which tops the in-flight set back up
+// to `parallelism`. This port expresses the identical rolling concurrency as
+// `parallelism` worker coroutines that each pull the next-closest uncontacted
+// node from the shared frontier under the session mutex, so at most
+// `parallelism` requests run at once and each completion immediately refills.
+// The TS Gate/withTimeout pair becomes structured concurrency: findClosestNodes
+// awaits every worker under a single folly timeout (a `done` flag, guarded by
+// the mutex, stands in for the single-shot Gate). Because the workers are
+// awaited rather than detached, `this` outlives them and no self-pin is needed.
+//
+// Discovery is I/O-bound (it spends its time awaiting RPC responses, not
+// computing), so unlike the Router's routing-table work it is not offloaded to
+// a dedicated worker executor; the fan-out runs on whatever executor drives the
+// join and yields at every RPC await.
+module;
+
+#include <chrono>
+#include <cstddef>
+#include <functional>
+#include <memory>
+#include <mutex>
+#include <optional>
+#include <set>
+#include <string>
+#include <utility>
+#include <vector>
+#include "packages/dht/protos/DhtRpc.pb.h"
+
+#include <coroutine> // IWYU pragma: keep
+
+export module streamr.dht.DiscoverySession;
+
+import streamr.utils.AbortController;
+import streamr.utils.CoroutineHelper;
+import streamr.utils.Uuid;
+import streamr.logger.SLogger;
+import streamr.dht.DhtNodeRpcRemote;
+import streamr.dht.getClosestNodes;
+import streamr.dht.getPeerDistance;
+import streamr.dht.Identifiers;
+import streamr.dht.PeerManager;
+
+// Hoisted from the former header (file scope, NOT exported).
+using streamr::logger::SLogger;
+using streamr::utils::AbortSignal;
+using streamr::utils::Uuid;
+
+export namespace streamr::dht::discovery {
+
+using ::dht::PeerDescriptor;
+using streamr::dht::DhtAddress;
+using streamr::dht::DhtAddressRaw;
+using streamr::dht::DhtNodeRpcRemote;
+using streamr::dht::Identifiers;
+using streamr::dht::PeerManager;
+using streamr::dht::contact::getClosestNodes;
+using streamr::dht::contact::GetClosestNodesOptions;
+using streamr::dht::helpers::getPeerDistance;
+
+struct DiscoverySessionOptions {
+    DhtAddress targetId;
+    size_t parallelism;
+    size_t noProgressLimit;
+    PeerManager& peerManager;
+    // Mutated by this session (and, when several entry points are joined at
+    // once, by the sibling sessions that share the same set). Owned by the
+    // caller; it outlives findClosestNodes.
+    std::set<DhtAddress>& contactedPeers;
+    AbortSignal& abortSignal;
+    std::function<std::shared_ptr<DhtNodeRpcRemote>(const PeerDescriptor&)>
+        createDhtNodeRpcRemote;
+};
+
+class DiscoverySession {
+private:
+    std::string id = Uuid::v4();
+    size_t noProgressCounter = 0;
+    std::set<DhtAddress> ongoingRequests;
+    bool done = false; // the single-shot Gate, guarded by the mutex
+    std::recursive_mutex mutex;
+    DiscoverySessionOptions options;
+
+    void addContacts(const std::vector<PeerDescriptor>& contacts) {
+        std::scoped_lock lock(this->mutex);
+        if (this->options.abortSignal.aborted || this->done) {
+            return;
+        }
+        for (const auto& contact : contacts) {
+            this->options.peerManager.addContact(contact);
+        }
+    }
+
+    folly::coro::Task<std::vector<PeerDescriptor>>
+    fetchClosestNeighborsFromRemote(PeerDescriptor peerDescriptor) {
+        {
+            std::scoped_lock lock(this->mutex);
+            if (this->options.abortSignal.aborted || this->done) {
+                co_return std::vector<PeerDescriptor>{};
+            }
+        }
+        const auto nodeId =
+            Identifiers::getNodeIdFromPeerDescriptor(peerDescriptor);
+        SLogger::trace("Getting closest neighbors from remote: " + nodeId);
+        const auto remote =
+            this->options.createDhtNodeRpcRemote(peerDescriptor);
+        auto returnedContacts =
+            co_await remote->getClosestPeers(this->options.targetId);
+        {
+            std::scoped_lock lock(this->mutex);
+            this->options.peerManager.setContactActive(nodeId);
+        }
+        co_return returnedContacts;
+    }
+
+    // Returns nullopt when there are no neighbours yet (TS would index [0] of
+    // an empty array; guarding avoids the undefined access).
+    [[nodiscard]] std::optional<PeerDescriptor> getClosestNeighbor() {
+        std::vector<PeerDescriptor> neighborDescriptors;
+        const auto neighbors = this->options.peerManager.getNeighbors();
+        neighborDescriptors.reserve(neighbors.size());
+        for (const auto& neighbor : neighbors) {
+            neighborDescriptors.push_back(neighbor->getPeerDescriptor());
+        }
+        const auto closest = getClosestNodes(
+            this->options.targetId,
+            neighborDescriptors,
+            GetClosestNodesOptions{.maxCount = 1});
+        if (closest.empty()) {
+            return std::nullopt;
+        }
+        return closest.front();
+    }
+
+    void onRequestSucceeded(
+        const DhtAddress& nodeId, const std::vector<PeerDescriptor>& contacts) {
+        std::scoped_lock lock(this->mutex);
+        if (!this->ongoingRequests.contains(nodeId)) {
+            return;
+        }
+        this->ongoingRequests.erase(nodeId);
+        const auto targetIdRaw =
+            Identifiers::getRawFromDhtAddress(this->options.targetId);
+        const auto oldClosestNeighbor = this->getClosestNeighbor();
+        this->addContacts(contacts);
+        const auto newClosestNeighbor = this->getClosestNeighbor();
+        // No prior neighbour means the newly added contacts are progress; a
+        // new closest that is not nearer than the old one is not.
+        if (oldClosestNeighbor.has_value() && newClosestNeighbor.has_value()) {
+            const auto oldClosestDistance = getPeerDistance(
+                targetIdRaw, DhtAddressRaw{oldClosestNeighbor->nodeid()});
+            const auto newClosestDistance = getPeerDistance(
+                targetIdRaw, DhtAddressRaw{newClosestNeighbor->nodeid()});
+            if (newClosestDistance >= oldClosestDistance) {
+                this->noProgressCounter++;
+            }
+        } else if (!newClosestNeighbor.has_value()) {
+            this->noProgressCounter++;
+        }
+    }
+
+    void onRequestFailed(const DhtAddress& nodeId) {
+        std::scoped_lock lock(this->mutex);
+        if (!this->ongoingRequests.contains(nodeId)) {
+            return;
+        }
+        this->ongoingRequests.erase(nodeId);
+        this->options.peerManager.removeContact(nodeId);
+    }
+
+    // One worker of the rolling fan-out: repeatedly claim the closest
+    // uncontacted node and query it, until the session is done.
+    folly::coro::Task<void> worker() {
+        while (true) {
+            std::optional<PeerDescriptor> node;
+            {
+                std::scoped_lock lock(this->mutex);
+                if (this->options.abortSignal.aborted || this->done) {
+                    co_return;
+                }
+                if (this->noProgressCounter >= this->options.noProgressLimit) {
+                    this->done = true;
+                    co_return;
+                }
+                const auto uncontacted = getClosestNodes(
+                    this->options.targetId,
+                    this->options.peerManager.getNearbyContacts(),
+                    GetClosestNodesOptions{
+                        .maxCount = this->options.parallelism,
+                        .excludedNodeIds = this->options.contactedPeers});
+                for (const auto& candidate : uncontacted) {
+                    const auto candidateId =
+                        Identifiers::getNodeIdFromPeerDescriptor(candidate);
+                    if (!this->ongoingRequests.contains(candidateId)) {
+                        node = candidate;
+                        break;
+                    }
+                }
+                if (!node.has_value()) {
+                    if (this->ongoingRequests.empty()) {
+                        this->done = true;
+                    }
+                    co_return;
+                }
+                const auto nodeId =
+                    Identifiers::getNodeIdFromPeerDescriptor(*node);
+                this->ongoingRequests.insert(nodeId);
+                // TS adds to contactedPeers synchronously at the start of
+                // fetchClosestNeighborsFromRemote, before the await; claiming
+                // it here (under the lock) keeps sibling workers from picking
+                // the same node.
+                this->options.contactedPeers.insert(nodeId);
+            }
+            const auto nodeId = Identifiers::getNodeIdFromPeerDescriptor(*node);
+            try {
+                const auto contacts =
+                    co_await this->fetchClosestNeighborsFromRemote(*node);
+                this->onRequestSucceeded(nodeId, contacts);
+            } catch (const std::exception& e) {
+                SLogger::trace(
+                    "getClosestPeers failed: " + std::string(e.what()));
+                this->onRequestFailed(nodeId);
+            }
+        }
+    }
+
+public:
+    explicit DiscoverySession(DiscoverySessionOptions options)
+        : options(std::move(options)) {}
+
+    [[nodiscard]] const std::string& getId() const { return this->id; }
+
+    folly::coro::Task<void> findClosestNodes(
+        std::chrono::milliseconds timeout) {
+        {
+            std::scoped_lock lock(this->mutex);
+            if (this->options.peerManager.getNearbyContactCount(
+                    this->options.contactedPeers) == 0) {
+                co_return;
+            }
+        }
+        std::vector<folly::coro::Task<void>> workers;
+        workers.reserve(this->options.parallelism);
+        for (size_t i = 0; i < this->options.parallelism; ++i) {
+            workers.push_back(this->worker());
+        }
+        co_await streamr::utils::co_withCancellation(
+            this->options.abortSignal.getCancellationToken(),
+            folly::coro::timeout(
+                folly::coro::collectAllRange(std::move(workers)), timeout));
+    }
+};
+
+} // namespace streamr::dht::discovery