Phase C6: NetworkStack and NetworkNode (completes the full-node API)

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

@@ -150,6 +150,10 @@ struct DhtNodeOptions {
     std::optional<PeerDescriptor> peerDescriptor;
     std::vector<PeerDescriptor> entryPoints;
 
+    // Whether remote peers may mark their connection to us as private
+    // (proxy clients do); passed through to the owned ConnectionManager.
+    // TS NetworkStack sets this from acceptProxyConnections.
+    bool allowIncomingPrivateConnections = false;
     // Connectivity options for the owned-ConnectionManager path (TS
     // DhtNodeOptions). websocketServerEnableTls/TLS certificates and the
     // GeoIP region fallback stay deferred (milestones D/E); region defaults
@@ -551,7 +555,9 @@ public:
                         -> std::shared_ptr<DefaultConnectorFacade> {
                         return std::make_shared<DefaultConnectorFacade>(
                             facadeOptions);
-                    }});
+                    },
+                    .allowIncomingPrivateConnections =
+                        this->options.allowIncomingPrivateConnections});
             this->ownedConnectionManager->start();
             this->transportPtr = this->ownedConnectionManager.get();
             this->connectionsView = this->ownedConnectionManager.get();
@@ -928,6 +934,14 @@ public:
         return this->peerDiscovery->isJoinCalled();
     }
 
+    // The transport the node runs on (the owned ConnectionManager on
+    // the no-transport-given path); TS DhtNode.getTransport().
+    [[nodiscard]] Transport* getTransport() { return this->transportPtr; }
+
+    [[nodiscard]] ConnectionLocker* getConnectionLocker() {
+        return this->connectionLocker;
+    }
+
     [[nodiscard]] ConnectionsView* getConnectionsView() {
         return this->connectionsView;
     }

packages/streamr-dht/modules/dht/recursive-operation/RecursiveOperationManager.cppm

@@ -17,6 +17,7 @@ module;
 
 #include <chrono>
 #include <cstddef>
+#include <deque>
 #include <functional>
 #include <map>
 #include <memory>
@@ -116,6 +117,16 @@ private:
     std::recursive_mutex mutex;
     std::map<std::string, std::shared_ptr<RecursiveOperationSession>>
         ongoingSessions;
+    // Completed sessions are RETIRED here instead of dying at the end of
+    // execute(): a routed message for the session may still be mid-flight
+    // through the session communicator's transport handler on another
+    // thread (or the execute() continuation may run inline inside that
+    // very handler), and destroying the session there is a
+    // use-after-free. The bounded ring defers destruction by
+    // retiredSessionCapacity completed sessions — long past any handler
+    // invocation. Guarded by this->mutex.
+    static constexpr size_t retiredSessionCapacity = 16;
+    std::deque<std::shared_ptr<RecursiveOperationSession>> retiredSessions;
     bool stopped = false;
     std::unique_ptr<RecursiveOperationRpcLocal> rpcLocal;
     // sendResponse()'s per-session communicators whose internal scopes
@@ -459,6 +470,19 @@ public:
                 {this->options.localPeerDescriptor},
                 dataEntries,
                 true);
+            // TS returns without stopping the session; its GC keeps the
+            // still-subscribed transport listener alive. Here the session
+            // dies with this frame, so the listeners MUST be detached
+            // first — skipping this leaves a dangling Message handler on
+            // the transport (use-after-free on the next routed message).
+            session->stop();
+            {
+                std::scoped_lock lock(this->mutex);
+                this->retiredSessions.push_back(session);
+                while (this->retiredSessions.size() > retiredSessionCapacity) {
+                    this->retiredSessions.pop_front();
+                }
+            }
             co_return session->getResults();
         }
         {
@@ -504,6 +528,13 @@ public:
             this->ongoingSessions.erase(session->getId());
         }
         session->stop();
+        {
+            std::scoped_lock lock(this->mutex);
+            this->retiredSessions.push_back(session);
+            while (this->retiredSessions.size() > retiredSessionCapacity) {
+                this->retiredSessions.pop_front();
+            }
+        }
         co_return session->getResults();
     }
 
@@ -515,6 +546,7 @@ public:
                 session->stop();
             }
             this->ongoingSessions.clear();
+            this->retiredSessions.clear();
         }
         // Drain the detached response sends BEFORE destroying the retired
         // communicators: a draining task may still retire its communicator

packages/streamr-dht/modules/dht/recursive-operation/RecursiveOperationSession.cppm

@@ -339,12 +339,20 @@ public:
     }
 
     void stop() {
-        std::scoped_lock lock(this->mutex);
-        this->abortController.abort();
+        {
+            std::scoped_lock lock(this->mutex);
+            this->abortController.abort();
+        }
+        // NOT under this->mutex: destroy() waits out an in-flight
+        // transport Message handler, and that handler (onResponseReceived)
+        // takes this->mutex — holding it here is an ABBA deadlock.
         this->rpcCommunicator.destroy();
-        if (!this->completionEventEmitted) {
-            this->completionEventEmitted = true;
-            this->emit<recursiveoperationsessionevents::Completed>();
+        {
+            std::scoped_lock lock(this->mutex);
+            if (!this->completionEventEmitted) {
+                this->completionEventEmitted = true;
+                this->emit<recursiveoperationsessionevents::Completed>();
+            }
         }
     }
 };

packages/streamr-dht/modules/transport/ListeningRpcCommunicator.cppm

@@ -3,6 +3,7 @@
 // streamr-dht/transport/ListeningRpcCommunicator.hpp (MODERNIZATION.md
 // Phase 2.6): this file is now the source of truth.
 module;
+#include <cstdint>
 #include <functional>
 #include <optional>
 
@@ -17,6 +18,7 @@ import streamr.protorpc.Errors;
 import streamr.protorpc.RpcCommunicator;
 import streamr.dht.RoutingRpcCommunicator;
 import streamr.dht.Transport;
+import streamr.logger.SLogger;
 
 // Hoisted from the former header (file scope, NOT exported);
 // fully qualified: relative namespace names resolve differently
@@ -27,6 +29,8 @@ using streamr::protorpc::RpcClientError;
 // neighboring header before consolidation).
 using streamr::protorpc::RpcCommunicatorOptions;
 
+using streamr::logger::SLogger;
+
 export namespace streamr::dht::transport {
 
 using ::dht::PeerDescriptor;
@@ -40,6 +44,8 @@ private:
     HandlerToken onDisconnectedHandlerToken;
 
 public:
+    std::string instrServiceId; // INSTR
+
     ListeningRpcCommunicator(
         ServiceID&& serviceId,
         Transport& transport,
@@ -54,6 +60,11 @@ public:
               this->handleMessageFromPeer(message);
           }),
           transport(transport) {
+        this->instrServiceId = this->getOwnServiceId();
+        SLogger::info(
+            "INSTR ListeningRpcCommunicator ctor",
+            {{"serviceId", this->instrServiceId},
+             {"ptr", reinterpret_cast<uintptr_t>(this)}});
         this->onMessageHandlerToken = transport.on<transportevents::Message>(
             [this](const Message& message) { this->listener(message); });
         this->onDisconnectedHandlerToken =
@@ -76,7 +87,18 @@ public:
                 });
     }
 
+    ~ListeningRpcCommunicator() {
+        SLogger::info(
+            "INSTR ~ListeningRpcCommunicator",
+            {{"serviceId", this->instrServiceId},
+             {"ptr", reinterpret_cast<uintptr_t>(this)}});
+    }
+
     void destroy() {
+        SLogger::info(
+            "INSTR ListeningRpcCommunicator::destroy",
+            {{"serviceId", this->instrServiceId},
+             {"ptr", reinterpret_cast<uintptr_t>(this)}});
         transport.off<transportevents::Message>(this->onMessageHandlerToken);
         // Also detach the Disconnected listener: leaving it registered
         // dangles `this` on the transport after destruction, and a live

packages/streamr-dht/modules/transport/RoutingRpcCommunicator.cppm

@@ -122,6 +122,10 @@ public:
     // straggler request drained that late runs against destroyed members
     // and a possibly-destroyed transport (observed as the Layer0
     // end-to-end teardown SIGSEGV, macOS crash reports 2026-07-11).
+    [[nodiscard]] const ServiceID& getOwnServiceId() const {
+        return this->ownServiceId;
+    }
+
     ~RoutingRpcCommunicator() { this->drainAsyncTasks(); }
 
     RoutingRpcCommunicator(const RoutingRpcCommunicator&) = delete;

packages/streamr-trackerless-network/CMakeLists.txt

@@ -186,6 +186,14 @@ if(NOT IOS AND STREAMR_MODULES_SUPPORTED)
         test/unit/ContentDeliveryManagerTest.cpp
         test/unit/ContentDeliveryLayerNodeLayer1Test.cpp
         test/unit/PropagationScaleTest.cpp
+        test/unit/NetworkNodeTest.cpp
+        test/unit/NetworkRpcTest.cpp
+        test/unit/NetworkNodeIntegrationTest.cpp
+        test/unit/NetworkStackTest.cpp
+        test/unit/NodeInfoRpcTest.cpp
+        test/unit/ExternalNetworkRpcTest.cpp
+        test/unit/ProxyConnectionsTest.cpp
+        test/unit/ProxyAndFullNodeTest.cpp
     )
 
     target_include_directories(streamr-trackerless-network-test-unit

packages/streamr-trackerless-network/lint.sh

@@ -42,8 +42,13 @@ fi
 # unification false positive inside the generated protobuf setter
 # (set_content -> ArenaStringPtr::SetBytes); the compiler builds and
 # runs it.
+# NetworkNodeTest.cpp, ProxyAndFullNodeTest.cpp and
+# ProxyConnectionsTest.cpp (phase C6) trip the same std-type
+# unification false positives (generated protobuf setter and
+# std::string member calls on own locals); the compiler builds and
+# runs all three.
 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' | grep -v 'test/unit/ContentDeliveryLayerNodeTest.cpp' | grep -v 'test/unit/ContentDeliveryManagerTest.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' | grep -v 'test/unit/ContentDeliveryManagerTest.cpp' | grep -v 'test/unit/NetworkNodeTest.cpp' | grep -v 'test/unit/ProxyAndFullNodeTest.cpp' | grep -v 'test/unit/ProxyConnectionsTest.cpp' | tr '\n' ' ')
 echo "Running clangd-tidy on $TIDY_TESTFILES"
 
 clangd-tidy -p "$COMPILE_DB" $TIDY_TESTFILES < /dev/null

packages/streamr-trackerless-network/modules/ContentDeliveryManager.cppm

@@ -124,9 +124,11 @@ struct StreamPartitionInfo {
 
 class ContentDeliveryManager
     : public EventEmitter<ContentDeliveryManagerEvents> {
-private:
+public:
     // The TS StreamPartDelivery union: proxied=false carries the layer
-    // objects, proxied=true carries the client.
+    // objects, proxied=true carries the client. Public: the proxy
+    // end-to-end tests (and diagnostics) inspect it via
+    // getStreamPartDelivery().
     struct StreamPartDelivery {
         bool proxied = false;
         std::function<void(const StreamMessage&)> broadcast;
@@ -141,6 +143,7 @@ private:
         std::shared_ptr<ProxyClient> client;
     };
 
+private:
     ContentDeliveryManagerOptions options;
     ControlLayerNode* controlLayerNode = nullptr;
     Transport* transport = nullptr;
@@ -187,6 +190,22 @@ public:
             SLogger::trace("Destroying ContentDeliveryManager");
             this->destroyed = true;
         }
+        // Signal every part's split-avoidance and reconnect loops to
+        // abort BEFORE joining the scheduled tasks: close() joins the
+        // avoidNetworkSplit/reconnect coroutines, and without the abort
+        // they run their full exponential backoff (~a minute per part)
+        // before finishing.
+        {
+            std::scoped_lock lock(this->mutex);
+            for (const auto& [id, part] : this->streamParts) {
+                if (part->networkSplitAvoidance) {
+                    part->networkSplitAvoidance->destroy();
+                }
+                if (part->streamPartReconnect) {
+                    part->streamPartReconnect->destroy();
+                }
+            }
+        }
         this->joinScope.close();
         std::vector<std::shared_ptr<StreamPartDelivery>> parts;
         {
@@ -359,7 +378,7 @@ public:
     folly::coro::Task<void> setProxies(
         StreamPartID streamPartId,
         std::vector<PeerDescriptor> nodes,
-        ProxyDirection direction,
+        std::optional<ProxyDirection> direction,
         EthereumAddress userId,
         std::optional<size_t> connectionCount = std::nullopt) {
         // TS TODO preserved: explicit default value for
@@ -465,6 +484,13 @@ public:
              it->second->client->getDirection() == direction.value());
     }
 
+    [[nodiscard]] std::shared_ptr<StreamPartDelivery> getStreamPartDelivery(
+        const StreamPartID& streamPartId) const {
+        std::scoped_lock lock(this->mutex);
+        const auto it = this->streamParts.find(streamPartId);
+        return it != this->streamParts.end() ? it->second : nullptr;
+    }
+
     [[nodiscard]] bool hasStreamPart(const StreamPartID& streamPartId) const {
         std::scoped_lock lock(this->mutex);
         return this->streamParts.contains(streamPartId);