From 3c4f54e15e3855efa3d6e0d818d5e78eca20cded Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Fri, 10 Jul 2026 10:00:33 +0300 Subject: [PATCH] Phase AA: fix the three concurrent-connect defects behind the join flakiness (#70) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root-caused with runtime instrumentation on both peers (per-decision logs with node ids and pending-connection identities, captured from failing runs of NonEntryPointNodesCanJoin); three distinct defects, each proven by a cited log line before fixing: 1. ConnectionManager::send() closed the connector-shared pending connection when the offerer tie-break rejected a concurrent outgoing accept. The connectors deduplicate connects per target, so the k-bucket ping and a discovery query racing to a new peer share ONE PendingConnection; the loser's close(false) tore down the endpoint the winner had just created (endpoint down 1 ms after creation, the racing send failing SEND_FAILED, the peer dropped from the k-bucket) and discarded every message buffered on it. Now the rejected send falls through and uses 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 buffered outgoing messages into one flat byte vector and flushed them as ONE send, while the receiving onData() parses each delivered blob as ONE protobuf Message — two buffered messages merged into one corrupt message (captured: flush bytes=549 count=2 -> receiver blobBytes=549 msgBytes=295) and all but one RPC request silently vanished, timing out 5 s later on a healthy connection. The buffer now keeps one entry per message and flushes one send per message, as TS does. 3. Teardown hang: the detached k-bucket ping tasks pin PeerManager via sharedFromThis, so DhtNode destroying its shared_ptr does not end them, and ~RoutingRpcCommunicator joined its executor while a dangling ping still used it (proven by sampling a live hung run). The ping is now cancellable (co_withCancellation with PeerManager's abort token) and PeerManager::stop() drains pingExecutor after the abort, outside the mutex. Verification: 25/25 isolated runs green under a hang watchdog before instrumentation removal (12x NonEntryPointNodesCanJoin, 8x CanJoinEvenIfANodeIsOffline, 5x CanJoinSimultaneously) and 16/16 after; full unit (172) + integration (15) suites green. Both formerly quarantined join tests are re-enabled (in the phase A8 test file, which lands with A8). The plan document's phase AA section records the full diagnosis. Co-authored-by: Claude Fable 5 --- .../modules/connection/ConnectionManager.cppm | 25 ++++- .../endpoint/ConnectingEndpointState.cppm | 25 +++-- .../streamr-dht/modules/dht/PeerManager.cppm | 49 +++++++--- trackerless-network-completion-plan.md | 91 +++++++++---------- 4 files changed, 116 insertions(+), 74 deletions(-) diff --git a/packages/streamr-dht/modules/connection/ConnectionManager.cppm b/packages/streamr-dht/modules/connection/ConnectionManager.cppm index e4ae73cd..dd232e27 100644 --- a/packages/streamr-dht/modules/connection/ConnectionManager.cppm +++ b/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); diff --git a/packages/streamr-dht/modules/connection/endpoint/ConnectingEndpointState.cppm b/packages/streamr-dht/modules/connection/endpoint/ConnectingEndpointState.cppm index d04d0527..9f44a529 100644 --- a/packages/streamr-dht/modules/connection/endpoint/ConnectingEndpointState.cppm +++ b/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 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> buffer; std::shared_ptr 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& 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) 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); diff --git a/packages/streamr-dht/modules/dht/PeerManager.cppm b/packages/streamr-dht/modules/dht/PeerManager.cppm index afb463dd..4daeabe4 100644 --- a/packages/streamr-dht/modules/dht/PeerManager.cppm +++ b/packages/streamr-dht/modules/dht/PeerManager.cppm @@ -211,7 +211,16 @@ private: [self, contact, nodeId]() -> folly::coro::Task { 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 getClosestRingContactsTo( diff --git a/trackerless-network-completion-plan.md b/trackerless-network-completion-plan.md index 478c5e3e..add814da 100644 --- a/trackerless-network-completion-plan.md +++ b/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)