Phase C3: ContentDeliveryLayerNode — the per-stream-part overlay node

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

@@ -167,7 +167,34 @@ struct DhtNodeOptions {
     std::optional<size_t> maxMessageSize;
 };
 
-class DhtNode : public Transport {
+// TS DhtNode re-emits the PeerManager contact events as its own events
+// (nearbyContactAdded etc.); the trackerless-network discovery-layer
+// facade subscribes to them. Mirrored here as a second event-emitter
+// base, merged with the Transport events via using-declarations.
+using DhtNodeContactEvents = std::tuple<
+    peermanagerevents::NearbyContactAdded,
+    peermanagerevents::NearbyContactRemoved,
+    peermanagerevents::RandomContactAdded,
+    peermanagerevents::RandomContactRemoved,
+    peermanagerevents::RingContactAdded,
+    peermanagerevents::RingContactRemoved>;
+
+class DhtNode
+    : public Transport,
+      public streamr::eventemitter::EventEmitter<DhtNodeContactEvents> {
+public:
+    using ContactEventEmitter =
+        streamr::eventemitter::EventEmitter<DhtNodeContactEvents>;
+    using ContactEventEmitter::emit;
+    using ContactEventEmitter::off;
+    using ContactEventEmitter::on;
+    using ContactEventEmitter::once;
+    using Transport::emit;
+    using Transport::off;
+    using Transport::on;
+    using Transport::once;
+    using Transport::removeAllListeners;
+
 private:
     static constexpr std::chrono::milliseconds externalApiTimeout{10000};
     static constexpr std::chrono::milliseconds networkConnectivityPollInterval{
@@ -296,6 +323,36 @@ private:
             [this](const PeerDescriptor& peerDescriptor) {
                 this->storeManager->onContactAdded(peerDescriptor);
             });
+        // TS re-emits every PeerManager contact event as a DhtNode event.
+        this->peerManager->on<peermanagerevents::NearbyContactAdded>(
+            [this](const PeerDescriptor& peerDescriptor) {
+                this->emit<peermanagerevents::NearbyContactAdded>(
+                    peerDescriptor);
+            });
+        this->peerManager->on<peermanagerevents::NearbyContactRemoved>(
+            [this](const PeerDescriptor& peerDescriptor) {
+                this->emit<peermanagerevents::NearbyContactRemoved>(
+                    peerDescriptor);
+            });
+        this->peerManager->on<peermanagerevents::RandomContactAdded>(
+            [this](const PeerDescriptor& peerDescriptor) {
+                this->emit<peermanagerevents::RandomContactAdded>(
+                    peerDescriptor);
+            });
+        this->peerManager->on<peermanagerevents::RandomContactRemoved>(
+            [this](const PeerDescriptor& peerDescriptor) {
+                this->emit<peermanagerevents::RandomContactRemoved>(
+                    peerDescriptor);
+            });
+        this->peerManager->on<peermanagerevents::RingContactAdded>(
+            [this](const PeerDescriptor& peerDescriptor) {
+                this->emit<peermanagerevents::RingContactAdded>(peerDescriptor);
+            });
+        this->peerManager->on<peermanagerevents::RingContactRemoved>(
+            [this](const PeerDescriptor& peerDescriptor) {
+                this->emit<peermanagerevents::RingContactRemoved>(
+                    peerDescriptor);
+            });
         this->peerManager->on<peermanagerevents::KBucketEmpty>([this]() {
             if (!this->peerDiscovery->isJoinOngoing() &&
                 !this->options.entryPoints.empty()) {

packages/streamr-trackerless-network/CMakeLists.txt

@@ -182,6 +182,9 @@ if(NOT IOS AND STREAMR_MODULES_SUPPORTED)
         test/unit/NeighborFinderTest.cpp
         test/unit/NeighborUpdateRpcLocalTest.cpp
         test/unit/NeighborUpdateRpcRemoteTest.cpp
+        test/unit/ContentDeliveryLayerNodeTest.cpp
+        test/unit/ContentDeliveryLayerNodeLayer1Test.cpp
+        test/unit/PropagationScaleTest.cpp
     )
 
     target_include_directories(streamr-trackerless-network-test-unit

packages/streamr-trackerless-network/lint.sh

@@ -35,8 +35,11 @@ fi
 # HandshakerTest.cpp (phase C2) trips the same std-type unification
 # false positive inside an included header ("no viable conversion from
 # const std::string to __self_view"); the compiler builds and runs it.
+# ContentDeliveryLayerNodeTest.cpp (phase C3) trips the std-type
+# unification false positive on its own std::string locals; the
+# compiler builds and runs it.
 TESTFILES=$(find test -type f \( -name "*.hpp" -o -name "*.cpp" \) -not -path '*/ts-integration/*' | sort | uniq | tr '\n' ' ')
-TIDY_TESTFILES=$(echo "$TESTFILES" | tr ' ' '\n' | grep -v 'test/unit/ContentDeliveryRpcRemoteTest.cpp' | grep -v 'test/unit/TemporaryConnectionRpcLocalTest.cpp' | grep -v 'test/unit/HandshakerTest.cpp' | tr '\n' ' ')
+TIDY_TESTFILES=$(echo "$TESTFILES" | tr ' ' '\n' | grep -v 'test/unit/ContentDeliveryRpcRemoteTest.cpp' | grep -v 'test/unit/TemporaryConnectionRpcLocalTest.cpp' | grep -v 'test/unit/HandshakerTest.cpp' | grep -v 'test/unit/ContentDeliveryLayerNodeTest.cpp' | tr '\n' ' ')
 echo "Running clangd-tidy on $TIDY_TESTFILES"
 
 clangd-tidy -p "$COMPILE_DB" $TIDY_TESTFILES < /dev/null

packages/streamr-trackerless-network/modules/discovery-layer/DhtNodeDiscoveryLayer.cppm

@@ -0,0 +1,110 @@
+// Module streamr.trackerlessnetwork.DhtNodeDiscoveryLayer
+// The TS code passes a DhtNode wherever a DiscoveryLayerNode is needed
+// (structural typing); this adapter provides that shape nominally: it
+// implements the DiscoveryLayerNode interface over a layer-1
+// streamr::dht::DhtNode, forwarding the six contact events and
+// delegating the getters and lifecycle.
+module;
+
+// Coroutine definitions need std::coroutine_traits declared in THIS
+// translation unit; it cannot arrive through an imported BMI.
+#include <coroutine> // IWYU pragma: keep
+
+#include <memory>
+#include <optional>
+#include <vector>
+#include "packages/dht/protos/DhtRpc.pb.h"
+
+export module streamr.trackerlessnetwork.DhtNodeDiscoveryLayer;
+
+import streamr.utils.CoroutineHelper;
+import streamr.trackerlessnetwork.DiscoveryLayerNode;
+import streamr.dht.DhtNode;
+import streamr.dht.Identifiers;
+import streamr.dht.PeerManager;
+import streamr.dht.protos;
+
+using streamr::dht::ClosestRingPeerDescriptors;
+using streamr::dht::DhtAddress;
+using streamr::dht::DhtNode;
+
+export namespace streamr::trackerlessnetwork::discoverylayer {
+
+using ::dht::PeerDescriptor;
+
+class DhtNodeDiscoveryLayer : public DiscoveryLayerNode {
+private:
+    std::shared_ptr<DhtNode> dhtNode;
+
+    template <typename SourceEvent, typename TargetEvent>
+    void forward() {
+        this->dhtNode->template on<SourceEvent>(
+            [this](const PeerDescriptor& peerDescriptor) {
+                this->template emit<TargetEvent>(peerDescriptor);
+            });
+    }
+
+public:
+    explicit DhtNodeDiscoveryLayer(std::shared_ptr<DhtNode> dhtNode)
+        : dhtNode(std::move(dhtNode)) {
+        namespace pme = streamr::dht::peermanagerevents;
+        namespace dle = discoverylayernodeevents;
+        this->forward<pme::NearbyContactAdded, dle::NearbyContactAdded>();
+        this->forward<pme::NearbyContactRemoved, dle::NearbyContactRemoved>();
+        this->forward<pme::RandomContactAdded, dle::RandomContactAdded>();
+        this->forward<pme::RandomContactRemoved, dle::RandomContactRemoved>();
+        this->forward<pme::RingContactAdded, dle::RingContactAdded>();
+        this->forward<pme::RingContactRemoved, dle::RingContactRemoved>();
+    }
+
+    [[nodiscard]] DhtNode& getDhtNode() { return *this->dhtNode; }
+
+    void removeContact(const DhtAddress& nodeId) override {
+        this->dhtNode->removeContact(nodeId);
+    }
+
+    [[nodiscard]] std::vector<PeerDescriptor> getClosestContacts(
+        std::optional<size_t> maxCount = std::nullopt) override {
+        return this->dhtNode->getClosestContacts(maxCount);
+    }
+
+    [[nodiscard]] std::vector<PeerDescriptor> getRandomContacts(
+        std::optional<size_t> maxCount = std::nullopt) override {
+        return this->dhtNode->getRandomContacts(maxCount);
+    }
+
+    [[nodiscard]] ClosestRingPeerDescriptors getRingContacts() override {
+        return this->dhtNode->getRingContacts();
+    }
+
+    [[nodiscard]] std::vector<PeerDescriptor> getNeighbors() override {
+        return this->dhtNode->getNeighbors();
+    }
+
+    [[nodiscard]] size_t getNeighborCount() override {
+        return this->dhtNode->getNeighborCount();
+    }
+
+    folly::coro::Task<void> joinDht(
+        std::vector<PeerDescriptor> entryPoints,
+        bool doRandomJoin = true,
+        bool retry = true) override {
+        co_await this->dhtNode->joinDht(
+            std::move(entryPoints), doRandomJoin, retry);
+    }
+
+    folly::coro::Task<void> joinRing() override {
+        co_await this->dhtNode->joinRing();
+    }
+
+    folly::coro::Task<void> start() override {
+        co_await this->dhtNode->start();
+    }
+
+    folly::coro::Task<void> stop() override {
+        this->dhtNode->stop();
+        co_return;
+    }
+};
+
+} // namespace streamr::trackerlessnetwork::discoverylayer