From 319fb8f7bf881addd2054db97ce4c76e72a345ca Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Sun, 12 Jul 2026 04:25:00 +0300 Subject: [PATCH] Phase C7: inspection (Inspector, InspectSession) Ports from the pinned TS 103.8.0-rc.3 (af966cf03), taken out of order: C7 depends only on C1's temporary-connection RPC (plus ListeningRpc- Communicator / ConnectionLocker / waitForEvent), not on C2/C3, so it can land while neighbor discovery is still in review. - InspectSession (modules/logic/inspection/): tracks which messages seen during an inspection were also delivered by the inspected node and emits Done as soon as that node proves it forwards traffic. TS toUserId(publisherId) in the message key becomes lowercase hex via BinaryUtils. - Inspector: opens a temporary connection to the inspected node (default open/close = TemporaryConnectionRpcRemote + weak connection locks, overridable callbacks like the TS options), waits for the session's Done with the inspection timeout, and returns the TS verdict formula. C++ deviations: the session/inspector maps are mutex-guarded (TS is single-threaded; here markMessage arrives from delivery threads while inspect() reads counts) and Done is emitted outside the locks. Tests: unit/InspectSession.test.ts and unit/Inspector.test.ts ported (5 gtests; the TS Promise.all waitForEvent pairs become folly collectAll, the jest setTimeout a helper thread); stable over --gtest_repeat=3; package suite 33/33. integration/Inspect.test.ts and end-to-end/inspect.test.ts need NetworkStack/full node and move to C6/C8 (noted in the plan). Co-Authored-By: Claude Fable 5 --- .../CMakeLists.txt | 2 + .../logic/inspection/InspectSession.cppm | 104 +++++++++ .../modules/logic/inspection/Inspector.cppm | 201 ++++++++++++++++++ .../test/unit/InspectSessionTest.cpp | 106 +++++++++ .../test/unit/InspectorTest.cpp | 107 ++++++++++ trackerless-network-completion-plan.md | 10 + 6 files changed, 530 insertions(+) create mode 100644 packages/streamr-trackerless-network/modules/logic/inspection/InspectSession.cppm create mode 100644 packages/streamr-trackerless-network/modules/logic/inspection/Inspector.cppm create mode 100644 packages/streamr-trackerless-network/test/unit/InspectSessionTest.cpp create mode 100644 packages/streamr-trackerless-network/test/unit/InspectorTest.cpp diff --git a/packages/streamr-trackerless-network/CMakeLists.txt b/packages/streamr-trackerless-network/CMakeLists.txt index d20045c2..8d6c1b0a 100644 --- a/packages/streamr-trackerless-network/CMakeLists.txt +++ b/packages/streamr-trackerless-network/CMakeLists.txt @@ -158,6 +158,8 @@ if(NOT IOS AND STREAMR_MODULES_SUPPORTED) test/unit/PeerDescriptorStoreManagerTest.cpp test/unit/StreamPartNetworkSplitAvoidanceTest.cpp test/unit/StreamPartReconnectTest.cpp + test/unit/InspectSessionTest.cpp + test/unit/InspectorTest.cpp test/unit/ProxyClientTest.cpp test/unit/ProxyConnectionRpcLocalTest.cpp test/unit/ProxyConnectionRpcRemoteTest.cpp diff --git a/packages/streamr-trackerless-network/modules/logic/inspection/InspectSession.cppm b/packages/streamr-trackerless-network/modules/logic/inspection/InspectSession.cppm new file mode 100644 index 00000000..dfda1c68 --- /dev/null +++ b/packages/streamr-trackerless-network/modules/logic/inspection/InspectSession.cppm @@ -0,0 +1,104 @@ +// Module streamr.trackerlessnetwork.InspectSession +// Ported from packages/trackerless-network/src/content-delivery-layer/ +// inspection/InspectSession.ts (v103.8.0-rc.3): tracks which of the +// messages seen during an inspection were (also) delivered by the +// inspected node, and emits Done as soon as the inspected node proves it +// forwards traffic. (The C++ package keeps TS's content-delivery-layer +// classes under modules/logic/, matching the existing tree.) +module; + +#include +#include +#include +#include +#include +#include +#include + +export module streamr.trackerlessnetwork.InspectSession; + +import streamr.trackerlessnetwork.protos; + +import streamr.dht.Identifiers; +import streamr.eventemitter.EventEmitter; +import streamr.utils.BinaryUtils; + +export namespace streamr::trackerlessnetwork::inspection { + +using streamr::dht::DhtAddress; +using streamr::eventemitter::Event; +using streamr::eventemitter::EventEmitter; +using streamr::utils::BinaryUtils; + +namespace inspectsessionevents { + +struct Done : Event<> {}; + +} // namespace inspectsessionevents + +using InspectSessionEvents = std::tuple; + +struct InspectSessionOptions { + DhtAddress inspectedNode; +}; + +class InspectSession : public EventEmitter { +private: + // TS runs single-threaded; here markMessage() arrives from delivery + // threads while inspect() reads the counts, so the map is guarded. + // Done is emitted OUTSIDE the lock (listeners like waitForEvent take + // their own locks). + std::mutex mutex; + // Value: has the message been received from the inspected node. + std::map inspectionMessages; + DhtAddress inspectedNode; + + static std::string createMessageKey(const ::MessageID& messageId) { + // TS toUserId(publisherId) renders the bytes as lowercase hex. + return BinaryUtils::binaryStringToHex(messageId.publisherid()) + ":" + + messageId.messagechainid() + ":" + + std::to_string(messageId.timestamp()) + ":" + + std::to_string(messageId.sequencenumber()); + } + +public: + explicit InspectSession(InspectSessionOptions options) + : inspectedNode(std::move(options.inspectedNode)) {} + + void markMessage( + const DhtAddress& remoteNodeId, const ::MessageID& messageId) { + bool done = false; + { + std::scoped_lock lock(this->mutex); + const auto messageKey = createMessageKey(messageId); + const auto it = this->inspectionMessages.find(messageKey); + if (it == this->inspectionMessages.end()) { + this->inspectionMessages.emplace( + messageKey, remoteNodeId == this->inspectedNode); + } else if (!it->second && remoteNodeId == this->inspectedNode) { + done = true; + } else if (it->second) { + done = true; + } + } + if (done) { + this->emit(); + } + } + + [[nodiscard]] size_t getInspectedMessageCount() { + std::scoped_lock lock(this->mutex); + return this->inspectionMessages.size(); + } + + [[nodiscard]] bool onlyMarkedByInspectedNode() { + std::scoped_lock lock(this->mutex); + return std::ranges::all_of( + this->inspectionMessages, + [](const auto& entry) { return entry.second; }); + } + + void stop() { this->emit(); } +}; + +} // namespace streamr::trackerlessnetwork::inspection diff --git a/packages/streamr-trackerless-network/modules/logic/inspection/Inspector.cppm b/packages/streamr-trackerless-network/modules/logic/inspection/Inspector.cppm new file mode 100644 index 00000000..719fb1bb --- /dev/null +++ b/packages/streamr-trackerless-network/modules/logic/inspection/Inspector.cppm @@ -0,0 +1,201 @@ +// Module streamr.trackerlessnetwork.Inspector +// Ported from packages/trackerless-network/src/content-delivery-layer/ +// inspection/Inspector.ts (v103.8.0-rc.3): opens a temporary connection to +// a suspected-misbehaving node, runs an InspectSession over the stream +// part's traffic, and reports whether the node forwards messages. The +// open/close operations are overridable callbacks (TS options style) so +// the unit test can substitute fakes. +module; + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include // IWYU pragma: keep + +export module streamr.trackerlessnetwork.Inspector; + +import streamr.dht.protos; +import streamr.trackerlessnetwork.protos; + +import streamr.dht.ConnectionLocker; +// export import: LockID appears in this module's public callback +// signatures, so every consumer needs the declaration visible. +export import streamr.dht.ConnectionLockStates; +import streamr.dht.Identifiers; +import streamr.dht.ListeningRpcCommunicator; +import streamr.logger.SLogger; +import streamr.trackerlessnetwork.InspectSession; +import streamr.trackerlessnetwork.TemporaryConnectionRpcRemote; +import streamr.utils.CoroutineHelper; +import streamr.utils.StreamPartID; +import streamr.utils.waitForEvent; + +// Hoisted from the former-header idiom (file scope, NOT exported). +using streamr::logger::SLogger; + +export namespace streamr::trackerlessnetwork::inspection { + +using ::dht::PeerDescriptor; +using streamr::dht::DhtAddress; +using streamr::dht::Identifiers; +using streamr::dht::connection::ConnectionLocker; +using streamr::dht::connection::LockID; +using streamr::dht::transport::ListeningRpcCommunicator; +using streamr::trackerlessnetwork::TemporaryConnectionRpcClient; +using streamr::trackerlessnetwork::TemporaryConnectionRpcRemote; +using streamr::utils::StreamPartID; + +using InspectConnectionFn = + std::function(PeerDescriptor, LockID)>; + +struct InspectorOptions { + PeerDescriptor localPeerDescriptor; + StreamPartID streamPartId; + ListeningRpcCommunicator& rpcCommunicator; + ConnectionLocker& connectionLocker; + std::optional inspectionTimeout; + // Empty function = use the default temporary-connection RPC. + InspectConnectionFn openInspectConnection; + InspectConnectionFn closeInspectConnection; +}; + +class Inspector { +private: + static constexpr std::chrono::milliseconds defaultInspectionTimeout{ + 60 * 1000}; + + PeerDescriptor localPeerDescriptor; + StreamPartID streamPartId; + ListeningRpcCommunicator& rpcCommunicator; + ConnectionLocker& connectionLocker; + std::chrono::milliseconds inspectionTimeout; + InspectConnectionFn openInspectConnection; + InspectConnectionFn closeInspectConnection; + // TS runs single-threaded; here markMessage() arrives from delivery + // threads while inspect() mutates the map on the caller's thread. + std::mutex mutex; + std::map> sessions; + + folly::coro::Task defaultOpenInspectConnection( + PeerDescriptor peerDescriptor, LockID lockId) { + TemporaryConnectionRpcRemote rpcRemote( + this->localPeerDescriptor, + peerDescriptor, + TemporaryConnectionRpcClient(this->rpcCommunicator)); + co_await rpcRemote.openConnection(); + this->connectionLocker.weakLockConnection( + Identifiers::getNodeIdFromPeerDescriptor(peerDescriptor), lockId); + } + + folly::coro::Task defaultCloseInspectConnection( + PeerDescriptor peerDescriptor, LockID lockId) { + TemporaryConnectionRpcRemote rpcRemote( + this->localPeerDescriptor, + peerDescriptor, + TemporaryConnectionRpcClient(this->rpcCommunicator)); + co_await rpcRemote.closeConnection(); + this->connectionLocker.weakUnlockConnection( + Identifiers::getNodeIdFromPeerDescriptor(peerDescriptor), lockId); + } + +public: + explicit Inspector(InspectorOptions options) + : localPeerDescriptor(std::move(options.localPeerDescriptor)), + streamPartId(std::move(options.streamPartId)), + rpcCommunicator(options.rpcCommunicator), + connectionLocker(options.connectionLocker), + inspectionTimeout( + options.inspectionTimeout.value_or(defaultInspectionTimeout)), + openInspectConnection( + options.openInspectConnection + ? std::move(options.openInspectConnection) + : [this]( + PeerDescriptor peerDescriptor, + LockID lockId) -> folly::coro::Task { + co_return co_await this->defaultOpenInspectConnection( + std::move(peerDescriptor), std::move(lockId)); + }), + closeInspectConnection( + options.closeInspectConnection + ? std::move(options.closeInspectConnection) + : [this]( + PeerDescriptor peerDescriptor, + LockID lockId) -> folly::coro::Task { + co_return co_await this->defaultCloseInspectConnection( + std::move(peerDescriptor), std::move(lockId)); + }) {} + + folly::coro::Task inspect(PeerDescriptor peerDescriptor) { + const auto nodeId = + Identifiers::getNodeIdFromPeerDescriptor(peerDescriptor); + auto session = std::make_shared( + InspectSessionOptions{.inspectedNode = nodeId}); + const LockID lockId{"inspector-" + this->streamPartId}; + { + std::scoped_lock lock(this->mutex); + this->sessions[nodeId] = session; + } + co_await this->openInspectConnection(peerDescriptor, lockId); + bool success = false; + try { + co_await streamr::utils::waitForEvent( + session.get(), this->inspectionTimeout); + success = true; + } catch (const std::exception&) { + SLogger::trace("Inspect session timed out, removing"); + } + // TS finally block: + co_await this->closeInspectConnection(peerDescriptor, lockId); + { + std::scoped_lock lock(this->mutex); + this->sessions.erase(nodeId); + } + co_return success || session->getInspectedMessageCount() < 1 || + session->onlyMarkedByInspectedNode(); + } + + void markMessage(const DhtAddress& sender, const ::MessageID& messageId) { + // Sessions are marked outside the map lock: markMessage may emit + // Done into listeners that take their own locks. + std::vector> currentSessions; + { + std::scoped_lock lock(this->mutex); + currentSessions.reserve(this->sessions.size()); + for (const auto& [_, session] : this->sessions) { + currentSessions.push_back(session); + } + } + for (const auto& session : currentSessions) { + session->markMessage(sender, messageId); + } + } + + [[nodiscard]] bool isInspected(const DhtAddress& nodeId) { + std::scoped_lock lock(this->mutex); + return this->sessions.contains(nodeId); + } + + void stop() { + std::vector> currentSessions; + { + std::scoped_lock lock(this->mutex); + for (const auto& [_, session] : this->sessions) { + currentSessions.push_back(session); + } + this->sessions.clear(); + } + for (const auto& session : currentSessions) { + session->stop(); + } + } +}; + +} // namespace streamr::trackerlessnetwork::inspection diff --git a/packages/streamr-trackerless-network/test/unit/InspectSessionTest.cpp b/packages/streamr-trackerless-network/test/unit/InspectSessionTest.cpp new file mode 100644 index 00000000..34a6e1c3 --- /dev/null +++ b/packages/streamr-trackerless-network/test/unit/InspectSessionTest.cpp @@ -0,0 +1,106 @@ +// Ported from packages/trackerless-network/test/unit/ +// InspectSession.test.ts (v103.8.0-rc.3). The TS Promise.all([waitForEvent, +// markMessage]) pairs become folly collectAll: the waitForEvent coroutine +// registers its listener before suspending, then the mark task fires the +// event. +#include +#include +#include +#include +#include + +#include // IWYU pragma: keep + +import streamr.dht.Identifiers; +import streamr.trackerlessnetwork.InspectSession; +import streamr.trackerlessnetwork.protos; +import streamr.utils.CoroutineHelper; +import streamr.utils.waitForEvent; + +using streamr::dht::DhtAddress; +using streamr::dht::Identifiers; +using streamr::trackerlessnetwork::inspection::InspectSession; +using streamr::trackerlessnetwork::inspection::InspectSessionOptions; +using streamr::trackerlessnetwork::inspection::inspectsessionevents::Done; +using streamr::utils::blockingWait; +using streamr::utils::waitForEvent; + +namespace { + +constexpr auto eventTimeout = std::chrono::milliseconds(100); +constexpr int64_t testTimestamp = 12345; + +::MessageID createMessageId(const std::string& messageChainId) { + ::MessageID messageId; + messageId.set_streamid("stream"); + messageId.set_messagechainid(messageChainId); + messageId.set_streampartition(0); + messageId.set_sequencenumber(0); + messageId.set_timestamp(testTimestamp); + messageId.set_publisherid("publisherId"); + return messageId; +} + +} // namespace + +class InspectSessionTest : public ::testing::Test { +protected: + ::MessageID messageId1 = createMessageId("messageChain0"); + ::MessageID messageId2 = createMessageId("messageChain1"); + // Fresh per test like the TS beforeEach (gtest constructs a new + // fixture per test; DhtAddress has no default constructor). + DhtAddress inspectedNode = Identifiers::createRandomDhtAddress(); + DhtAddress anotherNode = Identifiers::createRandomDhtAddress(); + std::unique_ptr inspectSession; + + void SetUp() override { + this->inspectSession = std::make_unique( + InspectSessionOptions{.inspectedNode = this->inspectedNode}); + } + + void TearDown() override { this->inspectSession->stop(); } + + // TS Promise.all([waitForEvent(session, 'done', 100), markMessage(...)]). + void expectDoneWhenMarking( + const DhtAddress& remoteNodeId, const ::MessageID& messageId) { + blockingWait( + folly::coro::collectAll( + waitForEvent(this->inspectSession.get(), eventTimeout), + folly::coro::co_invoke( + [this, &remoteNodeId, &messageId]() + -> folly::coro::Task { + this->inspectSession->markMessage( + remoteNodeId, messageId); + co_return; + }))); + } +}; + +TEST_F(InspectSessionTest, ShouldMarkMessage) { + this->inspectSession->markMessage(this->inspectedNode, this->messageId1); + EXPECT_EQ(this->inspectSession->getInspectedMessageCount(), 1); + this->inspectSession->markMessage(this->inspectedNode, this->messageId2); + EXPECT_EQ(this->inspectSession->getInspectedMessageCount(), 2); +} + +TEST_F(InspectSessionTest, ShouldEmitDoneWhenInspectedNodeSendsSeenMessage) { + this->inspectSession->markMessage(this->anotherNode, this->messageId1); + this->expectDoneWhenMarking(this->inspectedNode, this->messageId1); + EXPECT_EQ(this->inspectSession->getInspectedMessageCount(), 1); +} + +TEST_F( + InspectSessionTest, + ShouldEmitDoneWhenAnotherNodeSendsMessageAfterInspectedNode) { + this->inspectSession->markMessage(this->inspectedNode, this->messageId1); + this->expectDoneWhenMarking(this->anotherNode, this->messageId1); + EXPECT_EQ(this->inspectSession->getInspectedMessageCount(), 1); +} + +TEST_F(InspectSessionTest, ShouldNotEmitDoneIfMessageIdsDoNotMatch) { + this->inspectSession->markMessage(this->inspectedNode, this->messageId1); + EXPECT_THROW( + this->expectDoneWhenMarking(this->anotherNode, this->messageId2), + std::exception); + EXPECT_EQ(this->inspectSession->getInspectedMessageCount(), 2); +} diff --git a/packages/streamr-trackerless-network/test/unit/InspectorTest.cpp b/packages/streamr-trackerless-network/test/unit/InspectorTest.cpp new file mode 100644 index 00000000..e7a3f071 --- /dev/null +++ b/packages/streamr-trackerless-network/test/unit/InspectorTest.cpp @@ -0,0 +1,107 @@ +// Ported from packages/trackerless-network/test/unit/Inspector.test.ts +// (v103.8.0-rc.3). The TS setTimeout(..., 250) that marks the messages +// mid-inspection becomes a helper thread; openInspectConnection is +// overridden with a counting fake like the jest mock, the close path runs +// the real temporary-connection notify over a FakeTransport. +#include +#include +#include +#include +#include + +#include // IWYU pragma: keep + +import streamr.dht.FakeTransport; +import streamr.dht.Identifiers; +import streamr.dht.ListeningRpcCommunicator; +import streamr.dht.protos; +import streamr.trackerlessnetwork.Inspector; +import streamr.trackerlessnetwork.TestUtils; +import streamr.trackerlessnetwork.protos; +import streamr.utils.CoroutineHelper; +import streamr.utils.StreamPartID; + +using ::dht::PeerDescriptor; +using streamr::dht::DhtAddress; +using streamr::dht::Identifiers; +using streamr::dht::ServiceID; +using streamr::dht::connection::LockID; +using streamr::dht::transport::FakeTransport; +using streamr::dht::transport::ListeningRpcCommunicator; +using streamr::trackerlessnetwork::inspection::Inspector; +using streamr::trackerlessnetwork::inspection::InspectorOptions; +using streamr::trackerlessnetwork::testutils::createMockPeerDescriptor; +using streamr::trackerlessnetwork::testutils::MockConnectionLocker; +using streamr::utils::blockingWait; +using streamr::utils::StreamPartIDUtils; + +namespace { + +constexpr auto markDelay = std::chrono::milliseconds(250); +constexpr int64_t testTimestamp = 12345; + +::MessageID createMessageRef() { + ::MessageID messageId; + messageId.set_streamid("stream"); + messageId.set_messagechainid("messageChain0"); + messageId.set_streampartition(0); + messageId.set_sequencenumber(0); + messageId.set_timestamp(testTimestamp); + messageId.set_publisherid("publisher"); + return messageId; +} + +} // namespace + +class InspectorTest : public ::testing::Test { +protected: + PeerDescriptor inspectorDescriptor = createMockPeerDescriptor(); + PeerDescriptor inspectedDescriptor = createMockPeerDescriptor(); + DhtAddress nodeId = Identifiers::createRandomDhtAddress(); + ::MessageID messageRef = createMessageRef(); + std::atomic mockConnectCalls = 0; + FakeTransport transport{ + this->inspectorDescriptor, [](const auto& /*message*/) {}}; + ListeningRpcCommunicator rpcCommunicator{ + ServiceID{"inspector"}, this->transport}; + MockConnectionLocker connectionLocker; + std::unique_ptr inspector; + + void SetUp() override { + this->inspector = std::make_unique(InspectorOptions{ + .localPeerDescriptor = this->inspectorDescriptor, + .streamPartId = StreamPartIDUtils::parse("stream#0"), + .rpcCommunicator = this->rpcCommunicator, + .connectionLocker = this->connectionLocker, + .inspectionTimeout = std::nullopt, + .openInspectConnection = + [this](PeerDescriptor /*peerDescriptor*/, LockID /*lockId*/) + -> folly::coro::Task { + this->mockConnectCalls++; + co_return; + }, + .closeInspectConnection = {}}); + } + + void TearDown() override { this->inspector->stop(); } +}; + +TEST_F(InspectorTest, OpensInspectionConnectionAndRunsSuccessfully) { + std::thread marker([this]() { + std::this_thread::sleep_for(markDelay); + this->inspector->markMessage( + Identifiers::getNodeIdFromPeerDescriptor(this->inspectedDescriptor), + this->messageRef); + this->inspector->markMessage(this->nodeId, this->messageRef); + }); + const bool success = + blockingWait(this->inspector->inspect(this->inspectedDescriptor)); + marker.join(); + EXPECT_EQ(success, true); + EXPECT_EQ( + this->inspector->isInspected( + Identifiers::getNodeIdFromPeerDescriptor( + this->inspectedDescriptor)), + false); + EXPECT_EQ(this->mockConnectCalls, 1); +} diff --git a/trackerless-network-completion-plan.md b/trackerless-network-completion-plan.md index df922ea8..62b67843 100644 --- a/trackerless-network-completion-plan.md +++ b/trackerless-network-completion-plan.md @@ -602,6 +602,16 @@ decision 3.3.) *Tests to port:* `unit/InspectSession.test.ts`, `unit/Inspector.test.ts`, `integration/Inspect.test.ts`, `end-to-end/inspect.test.ts`. +*Implemented (phase-C7 PR, out of order — C7 has no C2/C3 dependency, only +C1's temporary-connection RPC):* both classes in `modules/logic/inspection/` +(namespace `...::inspection`; the package keeps TS's content-delivery-layer +classes under `modules/logic/`). Deviations: session/inspector maps are +mutex-guarded (markMessage arrives from delivery threads), Done is emitted +outside the locks; TS `toUserId(publisherId)` in the message key becomes +lowercase-hex via BinaryUtils. `integration/Inspect.test.ts` and +`end-to-end/inspect.test.ts` need `NetworkStack`/full node and move to +C6/C8. + **Phase C8 — Full-node end-to-end and TS interop.** *Tests to port:* `end-to-end/websocket-full-node-network.test.ts`, `end-to-end/webrtc-full-node-network.test.ts`,