Phase AA fixes to main (re-landing #70, which merged into its stack base)

packages/streamr-dht/modules/connection/ConnectionManager.cppm

@@ -371,15 +371,30 @@ public:
                 this->connectorFacade->createConnection(peerDescriptor);
             SLogger::debug("Created new connection");
             // This connection is outgoing (locally initiated). If the
-            // tie-break rejects it — a concurrent incoming connection from
-            // the same peer already won and owns the endpoint — discard the
-            // outgoing one we just created and fall through to that
-            // endpoint. (Simultaneous connect; see acceptNewConnection.)
+            // tie-break rejects it — an endpoint for this peer already exists
+            // (created by a concurrent send whose connect is in flight, or by
+            // the peer's incoming connection that won the simultaneous-connect
+            // tie-break) — do NOT close the pending connection here: the
+            // connector deduplicates per target, so `connection` may be the
+            // very pending connection the existing endpoint is currently
+            // connecting with, and closing it would tear that endpoint down
+            // (phase-AA bug: the k-bucket ping and a discovery query racing to
+            // the same new peer shared one pending connection; the loser's
+            // close churned the winner's endpoint and dropped its buffered
+            // messages). A genuinely losing duplicate is torn down by the
+            // protocol instead: the peer rejects its handshake with
+            // DUPLICATE_CONNECTION and the outgoing handshaker silences it
+            // (replaceAsDuplicate) — matching TS, which ignores the
+            // onNewConnection return value in send(). Fall through and use
+            // the existing endpoint.
             const bool accepted =
                 this->onNewConnection(connection, /*isLocalInitiated=*/true);
             SLogger::debug("Handled new connection");
             if (!accepted) {
-                connection->close(false);
+                SLogger::trace(
+                    "send(): outgoing connection rejected by the tie-break,"
+                    " using the existing endpoint for " +
+                    nodeId);
             }
             {
                 std::scoped_lock lock(this->endpointsMutex);

packages/streamr-dht/modules/connection/endpoint/ConnectingEndpointState.cppm

@@ -44,7 +44,13 @@ using streamr::dht::connection::IPendingConnection;
 class ConnectingEndpointState : public EndpointState {
 private:
     EndpointStateInterface& stateInterface;
-    std::vector<std::byte> buffer;
+    // One entry PER MESSAGE, flushed as one send() each: the receiving side
+    // parses every delivered blob as a single protobuf Message, so
+    // concatenating buffered messages into one flat byte vector (the previous
+    // implementation) merged them into one corrupt message on flush and
+    // silently destroyed all but the last RPC (phase-AA bug; TS buffers an
+    // array of messages the same way this does).
+    std::vector<std::vector<std::byte>> buffer;
     std::shared_ptr<IPendingConnection> pendingConnection;
     HandlerToken connectedHandlerToken;
     HandlerToken disconnectedHandlerToken;
@@ -193,11 +199,14 @@ public:
                         }
                         self->detachFromPendingConnection();
                         // Flushing under the mutex keeps the buffered
-                        // bytes ordered before any send() that observes
+                        // messages ordered before any send() that observes
                         // the connected state (connection->send() emits
-                        // nothing, so it is safe to call here).
+                        // nothing, so it is safe to call here). One send()
+                        // per message — boundaries must survive.
                         if (!self->buffer.empty()) {
-                            connection->send(self->buffer);
+                            for (const auto& message : self->buffer) {
+                                connection->send(message);
+                            }
                             self->buffer.clear();
                         }
                         self->stateInterface.enterConnectedState(connection);
@@ -243,7 +252,7 @@ public:
 
     void send(const std::vector<std::byte>& data) override {
         SLogger::debug("ConnectingEndpointState::send");
-        this->buffer.insert(this->buffer.end(), data.begin(), data.end());
+        this->buffer.push_back(data);
     }
 
     [[nodiscard]] DeferredCallout changeToConnectingState(
@@ -259,8 +268,12 @@ public:
         const std::shared_ptr<Connection>& connection) override {
         SLogger::debug("ConnectingEndpointState::changeToConnectedState");
         this->detachFromPendingConnection();
+        // One send() per message — boundaries must survive (see the buffer
+        // declaration).
         if (!this->buffer.empty()) {
-            connection->send(this->buffer);
+            for (const auto& message : this->buffer) {
+                connection->send(message);
+            }
             this->buffer.clear();
         }
         this->stateInterface.enterConnectedState(connection);

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

@@ -211,7 +211,16 @@ private:
                 [self, contact, nodeId]() -> folly::coro::Task<void> {
                     bool result = false;
                     try {
-                        result = co_await contact->ping();
+                        // Cancellable by stop(): a detached ping in flight at
+                        // teardown otherwise still references the owner's RPC
+                        // communicator while the owner is destroying it (the
+                        // phase-AA teardown hang: ~RpcCommunicatorClientApi
+                        // joins its executor while the dangling ping occupies
+                        // it). stop() aborts, then drains pingExecutor.
+                        result = co_await streamr::utils::co_withCancellation(
+                            self->abortController.getSignal()
+                                .getCancellationToken(),
+                            contact->ping());
                     } catch (const std::exception& err) {
                         SLogger::trace("ping threw " + std::string(err.what()));
                         result = false;
@@ -408,20 +417,32 @@ public:
     }
 
     void stop() {
-        std::scoped_lock lock(this->mutex);
-        this->stopped = true;
-        this->abortController.abort();
-        for (const auto& rpcRemote : this->neighbors->toArray()) {
-            rpcRemote->leaveNotice();
-            this->neighbors->remove(rpcRemote->getId());
-        }
-        this->neighbors->removeAllListeners();
-        for (const auto& rpcRemote : this->ringContacts->getAllContacts()) {
-            rpcRemote->leaveNotice();
-            this->ringContacts->removeContact(rpcRemote);
+        {
+            std::scoped_lock lock(this->mutex);
+            this->stopped = true;
+            this->abortController.abort();
+            for (const auto& rpcRemote : this->neighbors->toArray()) {
+                rpcRemote->leaveNotice();
+                this->neighbors->remove(rpcRemote->getId());
+            }
+            this->neighbors->removeAllListeners();
+            for (const auto& rpcRemote : this->ringContacts->getAllContacts()) {
+                rpcRemote->leaveNotice();
+                this->ringContacts->removeContact(rpcRemote);
+            }
+            this->nearbyContacts->stop();
+            this->randomContacts->stop();
         }
-        this->nearbyContacts->stop();
-        this->randomContacts->stop();
+        // Drain in-flight pings before returning (OUTSIDE the mutex: a
+        // draining ping's continuation takes it). The detached ping tasks
+        // pin `self`, so destroying the owner's shared_ptr does not end
+        // them — without this join a dangling ping still uses the RPC
+        // communicator while the owner destroys it (the phase-AA teardown
+        // hang). schedulePing enqueues under the mutex and onKBucketAdded
+        // checks `stopped`, so no new ping can be enqueued after the block
+        // above. The abort() cancels the in-flight RPCs, so the join
+        // returns promptly rather than waiting out the RPC timeout.
+        this->pingExecutor.join();
     }
 
     [[nodiscard]] RingContacts<DhtNodeRpcRemote> getClosestRingContactsTo(

trackerless-network-completion-plan.md

@@ -435,55 +435,48 @@ use a layer-0 `DhtNode` as its transport, sharing one `ConnectionManager` and co
 
 *Milestone A exit criterion:* a network of simulated C++ `DhtNode`s joins via entry points,
 routes messages, and stores/fetches data — the full dht test pyramid below the connector level
-is green, **except** the one test quarantined for phase AA below.
-
-**Phase AA — Concurrent-connect convergence (deferred; do AFTER the rest of Milestone A is
-merged and green).** Two multi-node join tests in `packages/streamr-dht/test/integration/
-MultipleEntryPointJoiningTest.cpp` are currently quarantined (`DISABLED_`), both failing on the
-same connection-establishment concurrency defect that only the threaded C++ runtime hits (TS is
-single-threaded and never does):
-- `DISABLED_NonEntryPointNodesCanJoin` — a fresh node joining a small already-formed
-  2-entry-point network (expects 3 neighbours); fails ~1/3 of runs.
-- `DISABLED_CanJoinEvenIfANodeIsOffline` — two nodes joining with one offline entry point
-  (expects 1 neighbour each). The `Simulator::removeConnector` fix cleanly fast-fails the offline
-  entry point, but the two *online* nodes still hit the same churn in ~20-30% of isolated runs
-  (fine in a warm back-to-back loop, but each CI run is a fresh isolated process); with only 1
-  expected neighbour there is no margin.
-
-`CanJoinSimultaneously` (3 nodes joining at once, 2 neighbours each) is **reliable** (8/8 isolated)
-and stays enabled — simultaneous joins connect symmetrically, so each pair is a genuine
-simultaneous connect the tie-break already converges.
-
-*Confirmed by traces:* the joining node discovers the earlier-joined peer (via an entry point's
-`getClosestPeers`), adds it to its k-bucket, then drops it — its `onKBucketAdded` ping to that
-peer fails with `SEND_FAILED: No connection to target, connection failed` and the handler
-`removeContact`s it. In the trace two **concurrent outgoing** connects to that same not-yet-
-connected peer are in flight (one from the PeerManager k-bucket ping on `pingExecutor`, one from
-the DiscoverySession `getClosestPeers` query on the join executor, on different threads); the
-endpoint is seen to adopt a pending connection then `detachFromPendingConnection` →
-`enterDisconnectedState` (the churn), erase itself, and the racing `send()` then finds no
-endpoint. `ConnectionManager::acceptNewConnection`'s offerer tie-break is built for a *simultaneous
-connect* (one incoming + one outgoing) and converges both sides order-independently on the
-offerer's connection, but it cannot disambiguate two **identical same-direction** duplicates —
-they are symmetric.
-
-*Two fixes were attempted and reverted (details in the session notes / auto-memory
-`blockingwait-cooperative-executor`):* (a) send()-level per-peer serialization of outgoing
-connects — **livelocked** under stress (a wait/retry cycle when a connection still churns; this
-also showed the churn is not fully explained by "two outgoing collide", since a deduped single
-outgoing should not churn yet the retry path still fired — the exact churn mechanism is not yet
-pinned down); (b) rejecting a same-direction duplicate in `acceptNewConnection` — **regressed**
-the test to always fail, because "first" is per-side (creation order at the initiator vs arrival
-order at the acceptor), so the two nodes settle on *different* connections and both fail.
-
-*Recommended approach for AA:* give each connection a **pair-stable id both peers can compare**
-(e.g. carry the initiator's connection/attempt id through the handshake to the acceptor, or derive
-one deterministically from the two node ids plus an exchanged nonce) and make the tie-break for
-two same-direction duplicates deterministically keep the same one on both sides (e.g. lowest id) —
-the only order-independent way for both peers to converge on the *same* duplicate. First reproduce
-and fully pin the churn mechanism (the open puzzle above) before committing to a design. Then
-re-enable the test and stress it (≥20 repeats) alongside the full connection integration suite.
-This is connection-layer (phase-A0) work independent of the DHT logic, hence deferred to here.
+is green.
+
+**Phase AA — Concurrent-connect convergence. RESOLVED (2026-07-10).** The multi-node join
+flakiness (`NonEntryPointNodesCanJoin` failing ~1/3 of isolated runs, `CanJoinEvenIfANodeIsOffline`
+~20-30%, plus occasional teardown hangs) was root-caused by runtime instrumentation on both peers
+(per-decision logs with node ids and pending-connection identities, captured from failing runs)
+and fixed. It was **three distinct defects**, none of them the previously-suspected
+"two symmetric duplicate connections" problem:
+
+1. *`ConnectionManager::send()` closed the connector-shared pending connection on a tie-break
+   rejection.* The connectors deduplicate outgoing connects per target (`connect()` to a peer with
+   a handshake already in flight returns the SAME `PendingConnection`), so when the k-bucket ping
+   and a discovery query raced to a new peer, both `send()`s held one shared pending connection;
+   the second `acceptNewConnection` was rejected by the offerer tie-break and `send()` then
+   `close(false)`d the shared connection — tearing down the endpoint the first send had just
+   created (log: `cache-HIT` same pc → `accept-REJECT` → `pc-close willEmitDisconnected=1` →
+   `endpoint-down` 1 ms after creation → ping `SEND_FAILED`). Each such churn also
+   `buffer.clear()`ed every other message buffered on the connecting endpoint (log: 5 requests
+   buffered, only the last delivered). **Fix:** never close on rejection — fall through and use
+   the existing endpoint; a genuinely losing duplicate is torn down by the DUPLICATE_CONNECTION
+   handshake protocol (`replaceAsDuplicate`), matching TS, which ignores the return value.
+2. *`ConnectingEndpointState` destroyed message boundaries on flush.* It buffered messages into
+   one flat byte vector and flushed them as ONE `connection->send()`, while the receiving
+   `onData()` parses each delivered blob as ONE protobuf `Message` — two buffered messages merged
+   into one corrupt message on the wire and all but one RPC request silently vanished (log:
+   `buffer-flush bytes=549 count=2` → receiver `blobBytes=549 msgBytes=295`, one message; the
+   lost ping then hit `RPC_TIMEOUT` after 5 s on a perfectly healthy connection). **Fix:** buffer
+   a vector of messages and flush one send per message (TS buffers an array the same way).
+3. *Teardown hang.* The detached k-bucket ping tasks pin `PeerManager` via `sharedFromThis`, so
+   destroying the owner's `shared_ptr` does not end them; `DhtNode` teardown then destroyed the
+   RPC communicator while a dangling ping still used it, and `~RpcCommunicatorClientApi` joined
+   its executor forever (proven by `sample` of a live hung process). **Fix:** the ping is wrapped
+   in `co_withCancellation` with `PeerManager`'s abort token, and `PeerManager::stop()` drains
+   `pingExecutor.join()` after the abort, outside the mutex.
+
+*Verification:* with all three fixes, 25/25 isolated runs green under a hang watchdog
+(12× NonEntryPointNodesCanJoin, 8× CanJoinEvenIfANodeIsOffline, 5× CanJoinSimultaneously), plus
+a post-cleanup stress round; both formerly quarantined tests are re-enabled; full unit +
+integration suites green. Two earlier model-driven fix attempts (send-level connect
+serialization; same-direction duplicate rejection) had failed because the real defects were in
+the connector dedup cache and the buffer framing — subsystems the hypotheses did not cover; the
+instrumentation found them from two captured failing runs.
 
 ### Milestone B — Direct connectivity (real sockets, NAT, WebRTC)