Phase C1: temporary-connection RPC, stream-part data keys, real C1 unit tests

packages/streamr-trackerless-network/CMakeLists.txt

@@ -43,6 +43,8 @@ find_package(Protobuf REQUIRED)
 find_package(Boost CONFIG REQUIRED)
 find_package(magic_enum CONFIG REQUIRED)
 find_package(folly REQUIRED)
+# streamPartIdToDataKey derives DHT data keys with SHA1 (OpenSSL EVP).
+find_package(OpenSSL REQUIRED)
 
 if(NOT TARGET Boost::uuid)
   add_library(Boost::uuid INTERFACE IMPORTED)
@@ -95,6 +97,7 @@ target_link_libraries(streamr-trackerless-network
     PUBLIC Boost::uuid
     PUBLIC magic_enum::magic_enum
     PUBLIC Folly::folly
+    PUBLIC OpenSSL::Crypto
     )
 
 export(TARGETS streamr-trackerless-network
@@ -113,6 +116,7 @@ file(WRITE "${CMAKE_BINARY_DIR}/streamr-trackerless-network-config.cmake"
     "find_package(Boost CONFIG REQUIRED)\n"
     "find_package(magic_enum CONFIG REQUIRED)\n"
     "find_package(folly REQUIRED)\n"
+    "find_package(OpenSSL REQUIRED)\n"
     "if(NOT TARGET Boost::uuid)\n"
     "  add_library(Boost::uuid INTERFACE IMPORTED)\n"
     "  set_target_properties(Boost::uuid PROPERTIES INTERFACE_INCLUDE_DIRECTORIES ${Boost_INCLUDE_DIRS})\n"
@@ -162,6 +166,9 @@ if(NOT IOS AND STREAMR_MODULES_SUPPORTED)
         test/unit/ContentDeliveryRpcRemoteTest.cpp
         test/unit/ContentDeliveryRpcLocalTest.cpp
         test/unit/DuplicateMessageDetectorTest.cpp
+        test/unit/NumberPairTest.cpp
+        test/unit/StreamPartIdToDataKeyTest.cpp
+        test/unit/TemporaryConnectionRpcLocalTest.cpp
     )
 
     target_include_directories(streamr-trackerless-network-test-unit
@@ -177,6 +184,11 @@ if(NOT IOS AND STREAMR_MODULES_SUPPORTED)
         PUBLIC streamr-trackerless-network
         PUBLIC streamr-trackerless-network-test-utils
         PUBLIC streamr::streamr-logger
+        # The C1 tests import streamr.dht.* and streamr.protorpc.*
+        # modules directly; imported modules are usable only by DIRECT
+        # consumers, so the owning targets are linked here.
+        PUBLIC streamr::streamr-dht
+        PUBLIC streamr::streamr-proto-rpc
         PUBLIC GTest::gtest
         streamr-trackerless-network-test-main
         PUBLIC Folly::folly

packages/streamr-trackerless-network/lint.sh

@@ -21,10 +21,22 @@ echo "Running clangd-tidy on $SRCFILES"
     ../../run-clang-format.py $SRCFILES
 fi
 
+# Two phase-C1 test files are excluded from clangd-tidy (owner-approved
+# selective disabling pattern, see packages/streamr-dht/lint.sh for the
+# full catalog of both false-positive classes):
+# ContentDeliveryRpcRemoteTest.cpp trips the global-namespace
+# duplicate-declaration merge failure (NetworkRpc/ProtoRpc types arrive
+# both textually and through imported BMIs; member calls like
+# requestid() are then diagnosed ambiguous), and
+# TemporaryConnectionRpcLocalTest.cpp trips the std-type unification
+# false positive (std::string-vs-std::string mismatch on its own
+# locals). The compiler builds and runs both (unit suite green on every
+# platform); clang-format still checks them.
 TESTFILES=$(find test -type f \( -name "*.hpp" -o -name "*.cpp" \) -not -path '*/ts-integration/*' | sort | uniq | tr '\n' ' ')
-echo "Running clangd-tidy on $TESTFILES"
+TIDY_TESTFILES=$(echo "$TESTFILES" | tr ' ' '\n' | grep -v 'test/unit/ContentDeliveryRpcRemoteTest.cpp' | grep -v 'test/unit/TemporaryConnectionRpcLocalTest.cpp' | tr '\n' ' ')
+echo "Running clangd-tidy on $TIDY_TESTFILES"
 
-clangd-tidy -p "$COMPILE_DB" $TESTFILES
+clangd-tidy -p "$COMPILE_DB" $TIDY_TESTFILES < /dev/null
 
 echo "Running clang-format --dry-run on $TESTFILES"
 ../../run-clang-format.py $TESTFILES

packages/streamr-trackerless-network/modules/logic/NodeList.cppm

@@ -118,9 +118,14 @@ public:
             std::ranges::to<std::vector<PeerDescriptor>>();
     }
 
+    // TS getNode() returns undefined for unknown ids; map::at would
+    // throw instead, so look up with find.
     [[nodiscard]] std::optional<std::shared_ptr<ContentDeliveryRpcRemote>> get(
         const DhtAddress& id) const {
-        return this->nodes.at(id);
+        if (const auto it = this->nodes.find(id); it != this->nodes.end()) {
+            return it->second;
+        }
+        return std::nullopt;
     }
 
     [[nodiscard]] size_t size(

packages/streamr-trackerless-network/modules/logic/streamPartIdToDataKey.cppm

@@ -0,0 +1,40 @@
+// Module streamr.trackerlessnetwork.streamPartIdToDataKey
+// Ported from streamPartIdToDataKey() in packages/trackerless-network/
+// src/ContentDeliveryManager.ts (v103.8.0-rc.3): stream-part entry
+// points are stored in the DHT under SHA1(streamPartId). The key must
+// match the TS derivation byte for byte, or C++ and TS nodes would
+// store/fetch the same stream part's entry points under different DHT
+// keys (wire compatibility — TypeScript wins).
+module;
+
+#include <string>
+#include <openssl/evp.h>
+
+export module streamr.trackerlessnetwork.streamPartIdToDataKey;
+
+import streamr.dht.Identifiers;
+import streamr.utils.StreamPartID;
+
+using streamr::dht::DhtAddress;
+using streamr::dht::DhtAddressRaw;
+using streamr::dht::Identifiers;
+using streamr::utils::StreamPartID;
+
+export namespace streamr::trackerlessnetwork {
+
+inline DhtAddress streamPartIdToDataKey(const StreamPartID& streamPartId) {
+    unsigned char digest[EVP_MAX_MD_SIZE]; // NOLINT
+    unsigned int digestLength = 0;
+    EVP_Digest(
+        streamPartId.data(),
+        streamPartId.size(),
+        digest, // NOLINT
+        &digestLength,
+        EVP_sha1(),
+        nullptr);
+    return Identifiers::getDhtAddressFromRaw(
+        DhtAddressRaw{std::string(
+            reinterpret_cast<const char*>(digest), digestLength)}); // NOLINT
+}
+
+} // namespace streamr::trackerlessnetwork

packages/streamr-trackerless-network/modules/logic/temporary-connection/TemporaryConnectionRpcLocal.cppm

@@ -0,0 +1,115 @@
+// Module streamr.trackerlessnetwork.TemporaryConnectionRpcLocal
+// Ported from packages/trackerless-network/src/content-delivery-layer/
+// temporary-connection/TemporaryConnectionRpcLocal.ts (v103.8.0-rc.3):
+// the server side of temporary content-delivery connections (used by
+// proxy/inspection flows), tracking the requesting nodes in a NodeList
+// and weak-locking the underlying connection while it is in use.
+module;
+
+#include <memory>
+#include <string>
+#include "packages/dht/protos/DhtRpc.pb.h"
+#include "packages/network/protos/NetworkRpc.pb.h"
+
+export module streamr.trackerlessnetwork.TemporaryConnectionRpcLocal;
+
+import streamr.trackerlessnetwork.NetworkRpcServer;
+import streamr.trackerlessnetwork.NetworkRpcClient;
+import streamr.trackerlessnetwork.ContentDeliveryRpcRemote;
+import streamr.trackerlessnetwork.NodeList;
+import streamr.dht.ConnectionLocker;
+import streamr.dht.ConnectionLockStates;
+import streamr.dht.DhtCallContext;
+import streamr.dht.Identifiers;
+import streamr.dht.ListeningRpcCommunicator;
+import streamr.dht.protos;
+import streamr.utils.StreamPartID;
+
+// Hoisted from the former header style (file scope, NOT exported);
+// fully qualified because relative namespace names resolve
+// differently at file scope than inside the package namespace.
+using streamr::dht::DhtAddress;
+using streamr::dht::Identifiers;
+using streamr::dht::connection::ConnectionLocker;
+using streamr::dht::connection::LockID;
+using streamr::dht::rpcprotocol::DhtCallContext;
+using streamr::dht::transport::ListeningRpcCommunicator;
+using streamr::utils::StreamPartID;
+
+export namespace streamr::trackerlessnetwork {
+
+using ::dht::PeerDescriptor;
+using TemporaryConnectionRpc =
+    streamr::protorpc::TemporaryConnectionRpc<DhtCallContext>;
+using ContentDeliveryRpcClient =
+    streamr::protorpc::ContentDeliveryRpcClient<DhtCallContext>;
+
+struct TemporaryConnectionRpcLocalOptions {
+    PeerDescriptor localPeerDescriptor;
+    StreamPartID streamPartId;
+    ListeningRpcCommunicator& rpcCommunicator;
+    ConnectionLocker& connectionLocker;
+};
+
+constexpr auto temporaryConnectionLockIdBase =
+    "system/content-delivery/temporary-connection/";
+
+class TemporaryConnectionRpcLocal : public TemporaryConnectionRpc {
+private:
+    // TS TODO preserved: use an options option or a named constant?
+    static constexpr size_t temporaryNodesLimit = 10;
+
+    TemporaryConnectionRpcLocalOptions options;
+    NodeList temporaryNodes;
+    LockID lockId;
+
+public:
+    explicit TemporaryConnectionRpcLocal(
+        TemporaryConnectionRpcLocalOptions options)
+        : options(std::move(options)),
+          temporaryNodes(
+              Identifiers::getNodeIdFromPeerDescriptor(
+                  this->options.localPeerDescriptor),
+              temporaryNodesLimit),
+          lockId(
+              LockID{
+                  temporaryConnectionLockIdBase + this->options.streamPartId}) {
+    }
+
+    [[nodiscard]] NodeList& getNodes() { return this->temporaryNodes; }
+
+    [[nodiscard]] bool hasNode(const DhtAddress& node) const {
+        return this->temporaryNodes.has(node);
+    }
+
+    void removeNode(const DhtAddress& nodeId) {
+        this->temporaryNodes.remove(nodeId);
+        this->options.connectionLocker.weakUnlockConnection(
+            nodeId, this->lockId);
+    }
+
+    TemporaryConnectionResponse openConnection(
+        const TemporaryConnectionRequest& /*request*/,
+        const DhtCallContext& context) override {
+        const auto& sender = context.incomingSourceDescriptor.value();
+        ContentDeliveryRpcClient client{this->options.rpcCommunicator};
+        const auto remote = std::make_shared<ContentDeliveryRpcRemote>(
+            this->options.localPeerDescriptor, sender, client);
+        this->temporaryNodes.add(remote);
+        this->options.connectionLocker.weakLockConnection(
+            Identifiers::getNodeIdFromPeerDescriptor(sender), this->lockId);
+        TemporaryConnectionResponse response;
+        response.set_accepted(true);
+        return response;
+    }
+
+    void closeConnection(
+        const CloseTemporaryConnection& /*request*/,
+        const DhtCallContext& context) override {
+        const auto remoteNodeId = Identifiers::getNodeIdFromPeerDescriptor(
+            context.incomingSourceDescriptor.value());
+        this->removeNode(remoteNodeId);
+    }
+};
+
+} // namespace streamr::trackerlessnetwork

packages/streamr-trackerless-network/modules/logic/temporary-connection/TemporaryConnectionRpcRemote.cppm

@@ -0,0 +1,88 @@
+// Module streamr.trackerlessnetwork.TemporaryConnectionRpcRemote
+// Ported from packages/trackerless-network/src/content-delivery-layer/
+// temporary-connection/TemporaryConnectionRpcRemote.ts (v103.8.0-rc.3):
+// the client side of temporary content-delivery connections. Both RPCs
+// swallow errors (open reports failure as `false`, close is
+// fire-and-forget), matching the TS behavior.
+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 <string>
+#include "packages/dht/protos/DhtRpc.pb.h"
+#include "packages/network/protos/NetworkRpc.pb.h"
+
+export module streamr.trackerlessnetwork.TemporaryConnectionRpcRemote;
+
+import streamr.utils.CoroutineHelper;
+import streamr.trackerlessnetwork.NetworkRpcClient;
+import streamr.dht.DhtCallContext;
+import streamr.dht.Identifiers;
+import streamr.dht.RpcRemote;
+import streamr.dht.protos;
+import streamr.logger.SLogger;
+
+// Hoisted (file scope, NOT exported); fully qualified because relative
+// namespace names resolve differently at file scope than inside the
+// package namespace.
+using streamr::dht::Identifiers;
+using streamr::dht::contact::RpcRemote;
+using streamr::dht::rpcprotocol::DhtCallContext;
+using streamr::logger::SLogger;
+
+export namespace streamr::trackerlessnetwork {
+
+using ::dht::PeerDescriptor;
+using TemporaryConnectionRpcClient =
+    streamr::protorpc::TemporaryConnectionRpcClient<DhtCallContext>;
+
+class TemporaryConnectionRpcRemote
+    : public RpcRemote<TemporaryConnectionRpcClient> {
+public:
+    TemporaryConnectionRpcRemote(
+        PeerDescriptor localPeerDescriptor, // NOLINT
+        PeerDescriptor remotePeerDescriptor,
+        TemporaryConnectionRpcClient client,
+        std::optional<std::chrono::milliseconds> timeout = std::nullopt)
+        : RpcRemote<TemporaryConnectionRpcClient>(
+              std::move(localPeerDescriptor),
+              std::move(remotePeerDescriptor),
+              client,
+              timeout) {}
+
+    folly::coro::Task<bool> openConnection() {
+        TemporaryConnectionRequest request;
+        auto options = this->formDhtRpcOptions({});
+        try {
+            const auto response = co_await this->getClient().openConnection(
+                std::move(request), std::move(options));
+            co_return response.accepted();
+        } catch (const std::exception& err) {
+            SLogger::debug(
+                "temporaryConnection to " +
+                Identifiers::getNodeIdFromPeerDescriptor(
+                    this->getPeerDescriptor()) +
+                " failed: " + std::string(err.what()));
+            co_return false;
+        }
+    }
+
+    folly::coro::Task<void> closeConnection() {
+        CloseTemporaryConnection request;
+        auto options = this->formDhtRpcOptions({});
+        try {
+            co_await this->getClient().closeConnection(
+                std::move(request), std::move(options));
+        } catch (const std::exception& err) {
+            SLogger::trace(
+                "closeConnection to " +
+                Identifiers::getNodeIdFromPeerDescriptor(
+                    this->getPeerDescriptor()) +
+                " failed: " + std::string(err.what()));
+        }
+    }
+};
+
+} // namespace streamr::trackerlessnetwork