Header graph → verified DAG: untangle the dht endpoint cluster (consolidation precondition 3)

MODERNIZATION.md

@@ -743,9 +743,17 @@ verified NDK clang (18 = r27), with the textual fallback below it.
    mis-unifies preamble/BMI types; with code in partitions there is no
    header fallback for the linter — coverage would regress from "full"
    to "none" for consolidated code).
-3. The dht intra-package include cycles (connection/endpoint cluster)
-   must become a partition DAG — untangle or use coarse per-cluster
-   partitions (deferred from 2.4 by design).
+3. ~~The dht intra-package include cycles (connection/endpoint cluster)
+   must become a partition DAG~~ **RESOLVED (post-2.6)** — a monorepo-wide
+   analysis (include edges + forward-declaration edges, which are what
+   break cycles textually but still force cyclic imports between
+   partitions) found exactly ONE cycle in all seven packages: the
+   6-header dht endpoint state-machine cluster. Untangled by making
+   `EndpointStateInterface` a pure abstract interface that `Endpoint`
+   implements (previously a concrete forwarder holding `Endpoint&` with
+   its member definitions at the bottom of Endpoint.hpp). Every
+   package's header graph is now a verified DAG, enforced continuously
+   by `check-include-dag.py` in `lint.sh`.
 
 ### What consolidation buys (quantified at the 2.4/2.5 checkpoints)
 - The −40% incremental-rebuild targets (currently: SLogger touch −6…−12%,
@@ -762,9 +770,10 @@ per-package with a grep-enforced "nothing includes them" gate.
 The façade stage is COMPLETE and delivers: uniform `import streamr.<pkg>`
 consumption, −24% clean builds, 250+ tests through import, and module
 infrastructure exercised on macOS/Linux/iOS/Android. Headers remain the
-linted source of truth. With the Android blocker resolved, consolidation
-is gated only by preconditions 2 (clangd lint coverage of purview code)
-and 3 (dht partition DAG). This is a stable resting point.
+linted source of truth. With the Android blocker resolved (2.6) and the
+header graph a verified DAG (post-2.6), consolidation is gated only by
+precondition 2: clangd lint coverage of purview code. This is a stable
+resting point.
 
 ## Success metrics (measured macOS + Linux, end of Phase 2.5)
 - ≥25% clean-build wall-clock reduction (dev build with tests).

check-include-dag.py

@@ -0,0 +1,132 @@
+#!/usr/bin/env python3
+"""Verify that each package's internal header graph is a DAG.
+
+The planned C++ modules consolidation (MODERNIZATION.md Phase 2.6) maps
+headers onto module partitions, and module imports must be acyclic — so
+the header dependency graph has to be a DAG. Two edge kinds are checked:
+
+  * include edges:   #include "streamr-<pkg>/..." between headers of the
+                     same package
+  * semantic edges:  a forward declaration (`class X;` / `struct X;`) of
+                     a type DEFINED in another header of the package.
+                     Forward declarations break cycles textually, but a
+                     forward-declared entity owned by another partition
+                     still forces a cyclic import after consolidation.
+
+Run from the repository root: ./check-include-dag.py
+Exits non-zero listing every cycle found.
+"""
+
+import os
+import re
+import sys
+from collections import defaultdict
+
+INC_RE = re.compile(r'#include\s+"((?:streamr|packages)[^"]*)"')
+FWD_RE = re.compile(r"^\s*(?:class|struct)\s+([A-Za-z_]\w*)\s*;")
+DEF_RE = re.compile(r"(?:class|struct)\s+([A-Za-z_]\w*)[^;{]*\{")
+
+
+def check_package(include_root: str) -> list[list[str]]:
+    nodes: set[str] = set()
+    texts: dict[str, str] = {}
+    for dirpath, _, files in os.walk(include_root):
+        for f in files:
+            if not f.endswith(".hpp"):
+                continue
+            rel = os.path.relpath(os.path.join(dirpath, f), include_root)
+            nodes.add(rel)
+            with open(
+                os.path.join(dirpath, f), encoding="utf-8", errors="replace"
+            ) as fh:
+                texts[rel] = fh.read()
+
+    owner: dict[str, str] = {}
+    for rel, txt in texts.items():
+        for m in DEF_RE.finditer(txt):
+            owner.setdefault(m.group(1), rel)
+
+    edges: dict[str, set[str]] = defaultdict(set)
+    for rel, txt in texts.items():
+        for line in txt.splitlines():
+            m = INC_RE.search(line)
+            if m and m.group(1) in nodes:
+                edges[rel].add(m.group(1))
+            fm = FWD_RE.match(line)
+            if fm:
+                own = owner.get(fm.group(1))
+                if own and own != rel:
+                    edges[rel].add(own)
+
+    # Tarjan SCC
+    sys.setrecursionlimit(10000)
+    counter = [0]
+    stack: list[str] = []
+    index: dict[str, int] = {}
+    lowlink: dict[str, int] = {}
+    on_stack: dict[str, bool] = {}
+    sccs: list[list[str]] = []
+
+    def strongconnect(v: str) -> None:
+        index[v] = lowlink[v] = counter[0]
+        counter[0] += 1
+        stack.append(v)
+        on_stack[v] = True
+        for w in edges.get(v, ()):
+            if w not in index:
+                strongconnect(w)
+                lowlink[v] = min(lowlink[v], lowlink[w])
+            elif on_stack.get(w):
+                lowlink[v] = min(lowlink[v], index[w])
+        if lowlink[v] == index[v]:
+            scc = []
+            while True:
+                w = stack.pop()
+                on_stack[w] = False
+                scc.append(w)
+                if w == v:
+                    break
+            sccs.append(scc)
+
+    for v in sorted(nodes):
+        if v not in index:
+            strongconnect(v)
+
+    return [s for s in sccs if len(s) > 1]
+
+
+def main() -> int:
+    packages_dir = "packages"
+    if not os.path.isdir(packages_dir):
+        print("check-include-dag.py: run from the repository root", file=sys.stderr)
+        return 2
+
+    failed = False
+    for pkg in sorted(os.listdir(packages_dir)):
+        include_root = os.path.join(packages_dir, pkg, "include")
+        if not os.path.isdir(include_root):
+            continue
+        cycles = check_package(include_root)
+        if cycles:
+            failed = True
+            print(f"{pkg}: header dependency cycles found:")
+            for scc in sorted(cycles, key=len, reverse=True):
+                print(f"  cycle of {len(scc)} headers:")
+                for h in sorted(scc):
+                    print(f"    {h}")
+        else:
+            print(f"{pkg}: OK")
+
+    if failed:
+        print(
+            "\nHeader cycles block the module consolidation "
+            "(see MODERNIZATION.md Phase 2.6). Break the cycle with an "
+            "abstract interface or by merging/splitting headers.",
+            file=sys.stderr,
+        )
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

lint.sh

@@ -9,6 +9,14 @@ if ! ./sync-cmake-files.sh --check; then
     exit 1
 fi
 
+# Verify that every package's internal header graph is a DAG (including
+# forward-declaration edges) — the module consolidation maps headers onto
+# partitions and module imports must be acyclic. See check-include-dag.py.
+if ! python3 ./check-include-dag.py; then
+    echo "::error title=header dependency cycle::A package's internal headers form a cycle; see the check-include-dag.py output above."
+    exit 1
+fi
+
 # Parse MonorepoPackages.cmake and loop through them
 for package in $(cat MonorepoPackages.cmake | grep -v "set(MonorepoPackages" | grep -v ")"); do
     echo ""

packages/streamr-dht/include/streamr-dht/connection/endpoint/Endpoint.hpp

@@ -39,22 +39,27 @@ using EndpointEvents = std::tuple<
     endpointevents::Connected,
     endpointevents::Disconnected>;
 
+// Endpoint implements EndpointStateInterface (privately) so the state
+// classes can drive the machine through the abstract interface — see
+// EndpointStateInterface.hpp for why this replaces the former concrete
+// forwarder member.
 class Endpoint : public EventEmitter<EndpointEvents>,
-                 public EnableSharedFromThis {
+                 public EnableSharedFromThis,
+                 private EndpointStateInterface {
 private:
     std::function<void()> removeSelfFromContainer;
     EndpointState* state;
     std::recursive_mutex mutex;
 
-    void emitData(const std::vector<std::byte>& data) {
+    void emitData(const std::vector<std::byte>& data) override {
         SLogger::debug("Endpoint::emitData start");
         auto self = sharedFromThis<Endpoint>();
         std::scoped_lock lock(this->mutex);
         this->emit<endpointevents::Data>(data);
         SLogger::debug("Endpoint::emitData end");
     }
 
-    void handleDisconnect(bool /*gracefulLeave*/) {
+    void handleDisconnect(bool /*gracefulLeave*/) override {
         SLogger::debug("Endpoint::handleDisconnect start");
         auto self = sharedFromThis<Endpoint>();
         std::scoped_lock lock(this->mutex);
@@ -65,7 +70,8 @@ class Endpoint : public EventEmitter<EndpointEvents>,
         SLogger::debug("Endpoint::handleDisconnect end");
     }
 
-    void changeToConnectedState(const std::shared_ptr<Connection>& connection) {
+    void changeToConnectedState(
+        const std::shared_ptr<Connection>& connection) override {
         SLogger::debug("Endpoint::changeToConnectedState start");
         auto self = sharedFromThis<Endpoint>();
         std::scoped_lock lock(this->mutex);
@@ -74,17 +80,14 @@ class Endpoint : public EventEmitter<EndpointEvents>,
         SLogger::debug("Endpoint::changeToConnectedState end");
     }
 
-    void handleConnected() {
+    void handleConnected() override {
         SLogger::debug("Endpoint::handleConnected start");
         auto self = sharedFromThis<Endpoint>();
         std::scoped_lock lock(this->mutex);
         this->emit<endpointevents::Connected>();
         SLogger::debug("Endpoint::handleConnected end");
     }
 
-    friend class EndpointStateInterface;
-
-    EndpointStateInterface stateInterface;
     std::shared_ptr<InitialEndpointState> initialState;
     std::shared_ptr<ConnectedEndpointState> connectedState;
     std::shared_ptr<ConnectingEndpointState> connectingState;
@@ -94,14 +97,10 @@ class Endpoint : public EventEmitter<EndpointEvents>,
     explicit Endpoint(
         PeerDescriptor peerDescriptor,
         std::function<void()>&& removeSelfFromContainer)
-        : stateInterface(*this),
-          connectedState(
-              ConnectedEndpointState::newInstance(this->stateInterface)),
-          connectingState(
-              ConnectingEndpointState::newInstance(this->stateInterface)),
-          disconnectedState(
-              DisconnectedEndpointState::newInstance(this->stateInterface)),
-          initialState(InitialEndpointState::newInstance(this->stateInterface)),
+        : connectedState(ConnectedEndpointState::newInstance(*this)),
+          connectingState(ConnectingEndpointState::newInstance(*this)),
+          disconnectedState(DisconnectedEndpointState::newInstance(*this)),
+          initialState(InitialEndpointState::newInstance(*this)),
           peerDescriptor(std::move(peerDescriptor)),
           removeSelfFromContainer(removeSelfFromContainer) {
         SLogger::debug("Endpoint constructor start");
@@ -128,7 +127,7 @@ class Endpoint : public EventEmitter<EndpointEvents>,
     ~Endpoint() override { SLogger::debug("Endpoint destructor"); }
 
     void changeToConnectingState(
-        const std::shared_ptr<IPendingConnection>& pendingConnection) {
+        const std::shared_ptr<IPendingConnection>& pendingConnection) override {
         SLogger::debug("Endpoint::changeToConnectingState start");
         auto self = sharedFromThis<Endpoint>();
         std::scoped_lock lock(this->mutex);
@@ -190,61 +189,6 @@ class Endpoint : public EventEmitter<EndpointEvents>,
     }
 };
 
-/*
-inline void ConnectedEndpointState::changeToConnectingState(
-    const std::shared_ptr<IPendingConnection>& pendingConnection) {
-    SLogger::debug("ConnectedEndpointState::changeToConnectingState start");
-    std::scoped_lock lock(this->connectedEndpointStateMutex);
-    this->connection->close(true);
-    this->removeEventHandlers();
-    this->stateInterface.changeToConnectingState(pendingConnection);
-    SLogger::debug("ConnectedEndpointState::changeToConnectingState end");
-}
-*/
-
-inline EndpointStateInterface::EndpointStateInterface(Endpoint& ep)
-    : endpoint(ep) {
-    SLogger::debug("EndpointStateInterface constructor");
-}
-
-inline void EndpointStateInterface::changeToConnectingState(
-    const std::shared_ptr<IPendingConnection>& pendingConnection) {
-    SLogger::debug("EndpointStateInterface::changeToConnectingState start");
-    this->endpoint.changeToConnectingState(pendingConnection);
-    SLogger::debug("EndpointStateInterface::changeToConnectingState end");
-}
-
-inline void EndpointStateInterface::changeToConnectedState(
-    const std::shared_ptr<Connection>& connection) {
-    SLogger::debug("EndpointStateInterface::changeToConnectedState start");
-    this->endpoint.changeToConnectedState(connection);
-    SLogger::debug("EndpointStateInterface::changeToConnectedState end");
-}
-
-inline void EndpointStateInterface::emitData(
-    const std::vector<std::byte>& data) {
-    SLogger::debug("EndpointStateInterface::emitData start");
-    this->endpoint.emitData(data);
-    SLogger::debug("EndpointStateInterface::emitData end");
-}
-
-inline void EndpointStateInterface::handleDisconnect(bool gracefulLeave) {
-    SLogger::debug("EndpointStateInterface::handleDisconnect start");
-    SLogger::debug(
-        "EndpointStateInterface::handleDisconnect acquiring scoped lock");
-    std::scoped_lock lock(this->endpoint.mutex);
-    SLogger::debug("EndpointStateInterface::handleDisconnect lock acquired");
-    this->endpoint.handleDisconnect(gracefulLeave);
-    SLogger::debug("EndpointStateInterface::handleDisconnect end");
-}
-
-inline void EndpointStateInterface::handleConnected() {
-    SLogger::debug("EndpointStateInterface::handleConnected start");
-
-    this->endpoint.handleConnected();
-    SLogger::debug("EndpointStateInterface::handleConnected end");
-}
-
 } // namespace streamr::dht::connection::endpoint
 
 #endif

packages/streamr-dht/include/streamr-dht/connection/endpoint/EndpointStateInterface.hpp

@@ -6,30 +6,32 @@
 #include <vector>
 #include "streamr-dht/connection/Connection.hpp"
 #include "streamr-dht/connection/IPendingConnection.hpp"
-#include "streamr-dht/connection/endpoint/EndpointState.hpp"
 
 namespace streamr::dht::connection::endpoint {
 
 using streamr::dht::connection::Connection;
 using streamr::dht::connection::IPendingConnection;
 
-class Endpoint;
-class EndpointState;
-
+// Pure abstract callback interface through which the endpoint state
+// classes drive the state machine. Endpoint implements it. Keeping this
+// abstract — instead of the former concrete forwarder that held an
+// Endpoint& and had its member definitions at the bottom of
+// Endpoint.hpp — makes the endpoint header cluster acyclic: Endpoint
+// depends on the states and the states depend only on this interface.
+// (This was the only header cycle in the monorepo; the planned module
+// consolidation needs the header graph to be a DAG.)
 class EndpointStateInterface {
-private:
-    Endpoint& endpoint;
-
 public:
-    explicit EndpointStateInterface(Endpoint& ep);
-    void changeToConnectingState(
-        const std::shared_ptr<IPendingConnection>& pendingConnection);
-    void changeToConnectedState(const std::shared_ptr<Connection>& connection);
-    void emitData(const std::vector<std::byte>& data);
-    void handleDisconnect(bool gracefulLeave);
-    void handleConnected();
+    virtual ~EndpointStateInterface() = default;
+    virtual void changeToConnectingState(
+        const std::shared_ptr<IPendingConnection>& pendingConnection) = 0;
+    virtual void changeToConnectedState(
+        const std::shared_ptr<Connection>& connection) = 0;
+    virtual void emitData(const std::vector<std::byte>& data) = 0;
+    virtual void handleDisconnect(bool gracefulLeave) = 0;
+    virtual void handleConnected() = 0;
 };
 
 } // namespace streamr::dht::connection::endpoint
 
-#endif // STREAMR_DHT_CONNECTION_ENDPOINT_ENDPOINTSTATEINTERFACE_HPP
+#endif // STREAMR_DHT_CONNECTION_ENDPOINT_ENDPOINTSTATEINTERFACE_HPP