From 16f728a505e5fdf890fdda8cea3e758015ae1d38 Mon Sep 17 00:00:00 2001 From: Amaury Chamayou Date: Fri, 28 Aug 2026 17:38:57 +0100 Subject: [PATCH 1/2] Resolve simultaneous node-to-node connects deterministically When two nodes dial each other at the same moment, each ends up holding both an outgoing connection it created and an incoming one the peer created. Both currently prefer the incoming connection, so each destroys the socket the other is relying on and both are left writing into a dead connection. On a healthy network this is masked, because closing a socket sends a FIN which the peer observes as a disconnect and repairs. It is not masked when that notification is lost, which is what a partition does. Both nodes now decide which connection to keep by ordering their node IDs: the lower ID keeps the one it opened, the higher prefers the one its peer opened. Both sides reach the same conclusion from information they already have, so exactly one connection survives and no orphan is created. This only applies to a recently created outgoing connection. An incoming connection from a peer we already have a settled connection to means that peer believes the link is broken, and it may have seen a failure we cannot see. Defending our own connection indefinitely would deny it the only means it has of repairing a link that is dead in a way we cannot detect. The host is never told its own node ID, but it appears in the sender field of every outbound message, and is only needed once an outgoing connection exists - which implies we have already sent. It is cached from there. NodeConnections is templated on its socket type, mirroring RPCConnections, so that the behaviour can be tested over a mock network which can suppress connection teardown the way a partition does. Production use is unchanged via NodeConnectionsImpl. Host connection lifecycle events are promoted from DEBUG to INFO. These are a handful of events per node lifetime, and their absence is what made the original failure require log reconstruction to diagnose. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- CMakeLists.txt | 6 + src/host/node_connections.h | 208 ++++++++--- src/host/test/node_connections.cpp | 552 +++++++++++++++++++++++++++++ 3 files changed, 722 insertions(+), 44 deletions(-) create mode 100644 src/host/test/node_connections.cpp diff --git a/CMakeLists.txt b/CMakeLists.txt index 5f075dd6a70b..c563ef27212b 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -712,6 +712,12 @@ if(BUILD_TESTS) ) target_link_libraries(rpc_connections_test PRIVATE uv) + add_unit_test( + node_connections_test + ${CMAKE_CURRENT_SOURCE_DIR}/src/host/test/node_connections.cpp + ) + target_link_libraries(node_connections_test PRIVATE uv) + add_unit_test( raft_test ${CMAKE_CURRENT_SOURCE_DIR}/src/consensus/aft/test/main.cpp diff --git a/src/host/node_connections.h b/src/host/node_connections.h index f929d189a9c1..82ec7cb53d59 100644 --- a/src/host/node_connections.h +++ b/src/host/node_connections.h @@ -14,23 +14,39 @@ namespace asynchost { static const auto UnassociatedNode = ccf::NodeId("Unknown"); - class NodeConnections + // See NodeConnectionsImpl::simultaneous_connect_window. + static constexpr auto default_simultaneous_connect_window = + std::chrono::seconds(2); + + template + class NodeConnectionsImpl { private: - class NodeConnectionBehaviour : public SocketBehaviour + // Identifies the current connection with a peer, whether we opened it, and + // when it was created. Knowing this is what allows two nodes which dialled + // each other at the same moment to agree on which connection to keep. + struct ConnectionInfo + { + ConnType socket = nullptr; + bool outgoing = false; + std::chrono::steady_clock::time_point created = + std::chrono::steady_clock::now(); + }; + + class NodeConnectionBehaviour : public SocketBehaviour { private: public: - NodeConnections& parent; + NodeConnectionsImpl& parent; std::optional node; std::optional msg_size = std::nullopt; std::vector pending; NodeConnectionBehaviour( const char* name, - NodeConnections& parent, + NodeConnectionsImpl& parent, std::optional node = std::nullopt) : - SocketBehaviour(name, "TCP"), + SocketBehaviour(name, "TCP"), parent(parent), node(std::move(node)) {} @@ -115,7 +131,13 @@ namespace asynchost if (!node.has_value()) { - associate_incoming(from); + if (!associate_incoming(from)) + { + // We are keeping an existing connection with this peer in + // preference to this one. Close it, rather than leaving two + // connections up and having to guess which one to write to. + return false; + } node = from; } @@ -147,23 +169,29 @@ namespace asynchost return true; } - virtual void associate_incoming(const ccf::NodeId& /*unused*/) {} + /// Returns false if this connection should be dropped rather than used. + virtual bool associate_incoming(const ccf::NodeId& /*unused*/) + { + return true; + } }; class NodeIncomingBehaviour : public NodeConnectionBehaviour { public: + using NodeConnectionBehaviour::parent; + size_t id; std::optional node_id; - NodeIncomingBehaviour(NodeConnections& parent, size_t id_) : + NodeIncomingBehaviour(NodeConnectionsImpl& parent, size_t id_) : NodeConnectionBehaviour("Node Incoming", parent), id(id_) {} void on_disconnect() override { - LOG_DEBUG_FMT("Disconnecting incoming connection {}", id); + LOG_INFO_FMT("Disconnecting incoming connection {}", id); parent.unassociated_incoming.erase(id); if (node_id.has_value()) @@ -172,10 +200,8 @@ namespace asynchost } } - void associate_incoming(const ccf::NodeId& n) override + bool associate_incoming(const ccf::NodeId& n) override { - node_id = n; - const auto unassociated = parent.unassociated_incoming.find(id); CCF_ASSERT_FMT( unassociated != parent.unassociated_incoming.end(), @@ -184,31 +210,60 @@ namespace asynchost n, id); - // Always prefer this (probably) newer connection. Pathological case is - // where both nodes open outgoings to each other at the same time, both - // see the corresponding incoming connections and _drop_ their outgoing - // connections. Both have a useless incoming connection they think they - // can use. Assumption is that they progress at different rates, and one - // of them eventually spots the dead connection and opens a new one - // which succeeds. - parent.connections[n] = unassociated->second; + // If we already have a *recently created* outgoing connection to this + // peer, then the two of us dialled each other at the same time and + // there are now two connections where one is needed. Both nodes must + // independently pick the same one to keep. If they do not, each + // destroys the connection the other is relying on, and both are left + // holding a socket whose far end is gone. Ordering by node ID gives + // both sides the same answer using only information they already have. + // + // The age check matters. An incoming connection from a peer we already + // have a settled connection to means that peer believes the link is + // broken, and it is in a better position to know - it may have seen an + // error we did not. Defending our own connection indefinitely would + // deny the peer the only means it has of repairing a link that is dead + // in a way we cannot detect, which is precisely the failure this change + // exists to prevent. So we only defend a connection young enough to + // still be part of a genuine race; anything older yields, as before. + const auto existing = parent.connections.find(n); + if ( + existing != parent.connections.end() && existing->second.outgoing && + (std::chrono::steady_clock::now() - existing->second.created) < + parent.simultaneous_connect_window && + !parent.prefer_incoming_from(n)) + { + LOG_INFO_FMT( + "Refusing incoming node connection ({}) from {}: keeping our " + "existing outgoing connection to resolve a simultaneous connect", + id, + n); + return false; + } + + node_id = n; + parent.connections[n] = {unassociated->second, false}; parent.unassociated_incoming.erase(unassociated); - LOG_DEBUG_FMT( - "Node incoming connection ({}) associated with {}", id, n); + LOG_INFO_FMT("Node incoming connection ({}) associated with {}", id, n); + return true; } }; class NodeOutgoingBehaviour : public NodeConnectionBehaviour { public: - NodeOutgoingBehaviour(NodeConnections& parent, const ccf::NodeId& node) : + using NodeConnectionBehaviour::node; + using NodeConnectionBehaviour::parent; + + NodeOutgoingBehaviour( + NodeConnectionsImpl& parent, const ccf::NodeId& node) : NodeConnectionBehaviour("Node Outgoing", parent, node) {} void on_bind_failed() override { - LOG_DEBUG_FMT( + LOG_INFO_FMT( "Disconnecting outgoing connection with {}: bind failed", *node); // NOLINT(bugprone-unchecked-optional-access) parent.remove_connection( @@ -217,7 +272,7 @@ namespace asynchost void on_resolve_failed() override { - LOG_DEBUG_FMT( + LOG_INFO_FMT( "Disconnecting outgoing connection with {}: resolve failed", *node); // NOLINT(bugprone-unchecked-optional-access) parent.remove_connection( @@ -226,7 +281,7 @@ namespace asynchost void on_connect_failed() override { - LOG_DEBUG_FMT( + LOG_INFO_FMT( "Disconnecting outgoing connection with {}: connect failed", *node); // NOLINT(bugprone-unchecked-optional-access) parent.remove_connection( @@ -235,7 +290,7 @@ namespace asynchost void on_disconnect() override { - LOG_DEBUG_FMT( + LOG_INFO_FMT( "Disconnecting outgoing connection with {}: disconnected", *node); // NOLINT(bugprone-unchecked-optional-access) parent.remove_connection( @@ -243,35 +298,49 @@ namespace asynchost } }; - class NodeServerBehaviour : public SocketBehaviour + class NodeServerBehaviour : public SocketBehaviour { public: - NodeConnections& parent; + NodeConnectionsImpl& parent; - NodeServerBehaviour(NodeConnections& parent) : - SocketBehaviour("Node Server", "TCP"), + NodeServerBehaviour(NodeConnectionsImpl& parent) : + SocketBehaviour("Node Server", "TCP"), parent(parent) {} - void on_accept(TCP& peer) override + void on_accept(ConnType& peer) override { auto id = parent.get_next_id(); peer->set_behaviour( std::make_unique(parent, id)); parent.unassociated_incoming.emplace(id, peer); - LOG_DEBUG_FMT("Accepted new incoming node connection ({})", id); + LOG_INFO_FMT("Accepted new incoming node connection ({})", id); } }; Ledger& ledger; - TCP listener; + ConnType listener; std::unordered_map> node_addresses; - std::unordered_map connections; + std::unordered_map connections; + + // How recently we must have opened an outgoing connection to treat an + // incoming connection from the same peer as a simultaneous connect rather + // than as that peer trying to repair a link it believes is broken. A real + // race is resolved within a round trip; this only needs to be long enough + // to cover one, and short enough that a peer's repair is not denied for + // any meaningful time. + std::chrono::milliseconds simultaneous_connect_window{ + default_simultaneous_connect_window}; + + // This node's own ID, which the host is never told directly. It is carried + // in the sender field of every outbound message, and is only needed once + // an outgoing connection exists - which implies we have already sent. + std::optional self_node_id = std::nullopt; - std::unordered_map unassociated_incoming; + std::unordered_map unassociated_incoming; size_t next_id = 1; ringbuffer::WriterPtr to_enclave; @@ -281,7 +350,7 @@ namespace asynchost std::nullopt; public: - NodeConnections( + NodeConnectionsImpl( messaging::Dispatcher& disp, Ledger& ledger, ringbuffer::AbstractWriterFactory& writer_factory, @@ -303,6 +372,13 @@ namespace asynchost register_message_handlers(disp); } + // Only used by tests, to exercise the behaviour either side of the window + // without having to wait for real time to pass. + void set_simultaneous_connect_window(std::chrono::milliseconds window) + { + simultaneous_connect_window = window; + } + void register_message_handlers( messaging::Dispatcher& disp) { @@ -331,7 +407,27 @@ namespace asynchost // Read piece-by-piece rather than all at once ccf::NodeId to = serialized::read(data, size); - TCP outbound_connection = nullptr; + // Peek at the sender ID without consuming it. This is the only place + // the host learns its own node ID, which is needed to resolve + // simultaneous connects consistently on both sides. + if (!self_node_id.has_value()) + { + const uint8_t* peek = data; + size_t peek_size = size; + try + { + serialized::read(peek, peek_size); + self_node_id = ccf::NodeId( + serialized::read(peek, peek_size)); + } + catch (const std::exception& e) + { + LOG_DEBUG_FMT( + "Unable to read own node ID from outbound: {}", e.what()); + } + } + + ConnType outbound_connection = nullptr; { const auto connection_it = connections.find(to); if (connection_it == connections.end()) @@ -355,7 +451,7 @@ namespace asynchost } else { - outbound_connection = connection_it->second; + outbound_connection = connection_it->second.socket; } } @@ -446,12 +542,33 @@ namespace asynchost } private: - TCP create_connection( + // Decide which of two simultaneously-created connections with a peer to + // keep. Both nodes evaluate this for the same pair and must agree: the node + // with the lower ID keeps the connection it opened, and the node with the + // higher ID prefers the one its peer opened. Exactly one of the two + // connections therefore survives, and both ends of it agree. + bool prefer_incoming_from(const ccf::NodeId& peer) const + { + if (!self_node_id.has_value()) + { + // We cannot have an outgoing connection without having sent a message, + // so this should be unreachable. Fall back to the previous behaviour of + // always preferring the newer connection. + LOG_FAIL_FMT( + "Resolving simultaneous connect with {} without knowing own node ID", + peer); + return true; + } + + return self_node_id.value().value() > peer.value(); + } + + ConnType create_connection( const ccf::NodeId& node_id, const std::string& host, const std::string& port) { - auto s = TCP(true, client_connection_timeout); + auto s = ConnType(true, client_connection_timeout); s->set_behaviour(std::make_unique(*this, node_id)); if (!s->connect(host, port, client_interface)) @@ -460,13 +577,14 @@ namespace asynchost return nullptr; } - connections.emplace(node_id, s); - LOG_DEBUG_FMT( + connections[node_id] = {s, true}; + LOG_INFO_FMT( "Added node connection with {} ({}:{})", node_id, host, port); return s; } + // Remove the connection with this peer, if any. bool remove_connection(const ccf::NodeId& node) { if (connections.erase(node) < 1) @@ -475,7 +593,7 @@ namespace asynchost return false; } - LOG_DEBUG_FMT("Removed node connection with {}", node); + LOG_INFO_FMT("Removed node connection with {}", node); return true; } @@ -491,4 +609,6 @@ namespace asynchost return id; } }; + + using NodeConnections = NodeConnectionsImpl; } diff --git a/src/host/test/node_connections.cpp b/src/host/test/node_connections.cpp new file mode 100644 index 000000000000..5859945716ea --- /dev/null +++ b/src/host/test/node_connections.cpp @@ -0,0 +1,552 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the Apache 2.0 License. + +#include "host/node_connections.h" + +#include "ds/messaging.h" +#include "ds/ring_buffer.h" + +#define DOCTEST_CONFIG_IMPLEMENT_WITH_MAIN +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace std::chrono_literals; + +// Surfaces the message of an unexpected exception in the failure output, +// rather than just reporting that something was thrown. +REGISTER_EXCEPTION_TRANSLATOR(const std::exception& e) +{ + return doctest::String(e.what()); +} + +namespace +{ + constexpr size_t buffer_size = 1 << 20; + + // Node IDs are hex-encoded public key hashes. The two chosen here sit at + // opposite ends of the ordering, so that any tie-break which depends on + // their relative order is unambiguous. + const std::string node_a_id(64, '0'); + const std::string node_b_id(64, 'f'); + + class MockSocketImpl; + + // Handle mirroring the subset of asynchost::TCP (proxy_ptr) that + // NodeConnections uses. Ownership is shared, so erasing an entry from the + // connections map destroys the underlying socket, exactly as proxy_ptr does. + class MockSocket + { + private: + std::shared_ptr impl; + + public: + MockSocket(); + MockSocket(std::nullptr_t) : impl(nullptr) {} + MockSocket(bool /*is_client*/, std::optional); + + MockSocketImpl* operator->() const + { + return impl.get(); + } + + [[nodiscard]] bool is_null() const + { + return impl == nullptr; + } + + [[nodiscard]] MockSocketImpl* get() const + { + return impl.get(); + } + }; + + // An in-process stand-in for the network. It gives the test explicit control + // over when connections complete, when bytes are delivered, and crucially + // whether the closure of one end of a connection is ever observed by the + // other end. + struct MockNet + { + struct PendingConnect + { + MockSocketImpl* initiator; + std::string host; + std::string port; + }; + + std::map listeners; + std::vector pending_connects; + std::set live_sockets; + // Which sockets each listener has accepted, so a test can reach the + // far end of a specific connection. + std::vector> accepted; + size_t next_port = 1000; + + // When false, destroying one end of a connection does not notify the other + // end. This models a link where the FIN is silently dropped rather than + // delivered: a firewall DROP rule, a NAT eviction, or a blackholing SDN. + // Without this, a dead socket is self-announcing and the bug under test + // cannot occur. + bool deliver_close = true; + + static std::string key(const std::string& host, const std::string& port) + { + return host + ":" + port; + } + }; + + MockNet* net = nullptr; + + class MockSocketImpl : public std::enable_shared_from_this + { + public: + std::unique_ptr> behaviour; + std::string host; + std::string port; + MockSocketImpl* peer = nullptr; + std::vector rx; + + // Writes issued before the connection completes are buffered and flushed + // on connect, as libuv does via TCPImpl::pending_writes. Without this a + // node's first message to a peer would be lost, which never happens in + // practice and would mask the behaviour under test. + bool connected = false; + std::vector pending_tx; + + MockSocketImpl() + { + net->live_sockets.insert(this); + } + + ~MockSocketImpl() + { + net->live_sockets.erase(this); + sever(); + } + + // Detach from the far end, notifying it only if the network is currently + // delivering connection teardown. + void sever() + { + if (peer != nullptr) + { + auto* p = peer; + peer = nullptr; + p->peer = nullptr; + if (net->deliver_close && p->behaviour != nullptr) + { + p->behaviour->on_disconnect(); + } + } + } + + void on_connected(MockSocketImpl* far_end) + { + peer = far_end; + connected = true; + if (!pending_tx.empty()) + { + peer->rx.insert(peer->rx.end(), pending_tx.begin(), pending_tx.end()); + pending_tx.clear(); + } + } + + void set_behaviour( + std::unique_ptr> b) + { + behaviour = std::move(b); + } + + [[nodiscard]] std::string get_host() const + { + return host; + } + + [[nodiscard]] std::string get_port() const + { + return port; + } + + bool listen( + const std::string& host_, + const std::string& port_, + const std::optional& = std::nullopt) + { + host = host_; + port = port_ == "0" ? std::to_string(net->next_port++) : port_; + net->listeners[MockNet::key(host, port)] = this; + return true; + } + + bool connect( + const std::string& host_, + const std::string& port_, + const std::optional& = std::nullopt) + { + net->pending_connects.push_back({this, host_, port_}); + return true; + } + + bool write(size_t len, const uint8_t* data, sockaddr = {}) + { + if (!connected) + { + pending_tx.insert(pending_tx.end(), data, data + len); + return true; + } + + if (peer == nullptr) + { + // Writing into a socket whose far end is gone. As with a real + // blackholed TCP connection this silently succeeds, and the bytes are + // never seen again. + return true; + } + + peer->rx.insert(peer->rx.end(), data, data + len); + return true; + } + }; + + MockSocket::MockSocket() : impl(std::make_shared()) {} + MockSocket::MockSocket(bool, std::optional) : + impl(std::make_shared()) + {} + + // Complete every outstanding connect at once. Batching them is what makes + // the simultaneous-connect race deterministic: both nodes have dialled each + // other before either has accepted. + void complete_connects() + { + auto pending = std::move(net->pending_connects); + net->pending_connects.clear(); + + for (const auto& p : pending) + { + auto listener = net->listeners.find(MockNet::key(p.host, p.port)); + REQUIRE(listener != net->listeners.end()); + + // The accepted socket is the far end of the initiator's socket. on_accept + // takes ownership of it, as NodeServerBehaviour does for a real peer. + auto accepted = MockSocket(); + accepted->connected = true; + accepted->peer = p.initiator; + + listener->second->behaviour->on_accept(accepted); + net->accepted.emplace_back(listener->second, accepted.get()); + p.initiator->on_connected(accepted.get()); + p.initiator->behaviour->on_connect(); + } + } + + // Deliver buffered bytes to the behaviour on each receiving socket, repeating + // until the network is quiescent. + void pump() + { + for (size_t round = 0; round < 8; ++round) + { + std::vector with_data; + for (auto* s : net->live_sockets) + { + if (!s->rx.empty() && s->behaviour != nullptr) + { + with_data.push_back(s); + } + } + + if (with_data.empty()) + { + return; + } + + for (auto* s : with_data) + { + // The socket may have been destroyed by an earlier delivery in this + // same round. + if (net->live_sockets.find(s) == net->live_sockets.end()) + { + continue; + } + + auto data = s->rx; + s->rx.clear(); + uint8_t* ptr = data.data(); + if (!s->behaviour->on_read(data.size(), ptr, {})) + { + // Returning false from on_read closes the connection, as it does for + // a real socket. Keep the socket alive across the callback, since + // its owner will typically drop it in response. + auto keep_alive = s->shared_from_this(); + s->sever(); + s->behaviour->on_disconnect(); + } + } + } + } + + // The host inspects the raft header of every outbound consensus message, so + // the payload has to be a plausible one. A pre-vote is used here because it + // is what a stalled candidate sends, and unlike an append-entries it carries + // no ledger indices for the host to read. + std::vector make_pre_vote() + { + std::vector m(sizeof(aft::Node2NodeMsg) + 16, 0); + const auto type = + static_cast(aft::raft_request_pre_vote); + std::memcpy(m.data(), &type, sizeof(type)); + return m; + } + + // The socket a given listener most recently accepted, or nullptr. + MockSocketImpl* last_accepted_by(MockSocketImpl* listener) + { + MockSocketImpl* found = nullptr; + for (const auto& [l, a] : net->accepted) + { + if (l == listener && net->live_sockets.count(a) > 0) + { + found = a; + } + } + return found; + } + + struct TestNode + { + ringbuffer::TestBuffer to_enclave; + ringbuffer::TestBuffer to_host; + ringbuffer::Circuit circuit; + ringbuffer::WriterFactory wf; + + // Processes messages the enclave sent to the host + messaging::BufferProcessor host_bp; + // Processes messages the host sent to the enclave + messaging::BufferProcessor enclave_bp; + + std::filesystem::path ledger_dir; + std::unique_ptr ledger; + std::unique_ptr> connections; + + ringbuffer::WriterPtr enclave_writer; + + std::string host; + std::string port; + size_t received = 0; + + TestNode(const std::string& name, std::string host_) : + to_enclave(buffer_size), + to_host(buffer_size), + circuit(to_enclave.bd, to_host.bd), + wf(circuit), + host_bp("node_host"), + enclave_bp("node_enclave"), + ledger_dir( + std::filesystem::temp_directory_path() / ("nc_test_ledger_" + name)), + host(std::move(host_)), + port("0") + { + std::filesystem::remove_all(ledger_dir); + ledger = std::make_unique(ledger_dir.string(), wf); + connections = + std::make_unique>( + host_bp.get_dispatcher(), *ledger, wf, host, port, std::nullopt, 2s); + + enclave_writer = wf.create_writer_to_outside(); + + DISPATCHER_SET_MESSAGE_HANDLER( + enclave_bp, + ccf::node_inbound, + [this](const uint8_t* data, size_t size) { + auto [msg_type, from, payload] = + ringbuffer::read_message(data, size); + (void)msg_type; + (void)from; + (void)payload; + ++received; + }); + } + + ~TestNode() + { + connections.reset(); + ledger.reset(); + std::filesystem::remove_all(ledger_dir); + } + + void drain_to_host() + { + host_bp.read_all(circuit.read_from_inside()); + } + + void drain_to_enclave() + { + enclave_bp.read_all(circuit.read_from_outside()); + } + + void send_to(const std::string& peer_id, const std::vector& body) + { + RINGBUFFER_WRITE_MESSAGE( + ccf::node_outbound, + enclave_writer, + peer_id, + ccf::NodeMsgType::consensus_msg, + self_id, + body); + } + + void learn_address( + const std::string& peer_id, + const std::string& peer_host, + const std::string& peer_port) + { + RINGBUFFER_WRITE_MESSAGE( + ccf::associate_node_address, + enclave_writer, + peer_id, + peer_host, + peer_port); + } + + std::string self_id; + }; +} + +// Regression test for the node-to-node channel stall investigated in #8232. +// +// When two nodes dial each other at the same moment, each ends up with both an +// outgoing connection it created and an incoming connection the peer created. +// If both nodes independently decide to keep the incoming one, they each +// destroy the socket the other is relying on, and both are left holding a +// connection whose far end no longer exists. +// +// On a healthy network this is self-correcting, because closing a socket sends +// a FIN which the peer observes as a disconnect. It is not self-correcting when +// that notification is lost, which is exactly what a partition does. The result +// is two nodes that believe they are connected, silently writing consensus +// messages into dead sockets indefinitely. +TEST_CASE("Simultaneous connect leaves a usable connection in both directions") +{ + MockNet mock_net; + net = &mock_net; + + { + TestNode a("a", "10.0.0.1"); + TestNode b("b", "10.0.0.2"); + a.self_id = node_a_id; + b.self_id = node_b_id; + + a.learn_address(node_b_id, b.host, b.port); + b.learn_address(node_a_id, a.host, a.port); + a.drain_to_host(); + b.drain_to_host(); + + const auto payload = make_pre_vote(); + + // Both nodes send to each other before either has accepted a connection. + // This is the simultaneous-connect race. + a.send_to(node_b_id, payload); + b.send_to(node_a_id, payload); + a.drain_to_host(); + b.drain_to_host(); + + // From here the link silently swallows connection teardown, so neither node + // can rely on TCP to tell it that a socket has become useless. + mock_net.deliver_close = false; + + complete_connects(); + pump(); + a.drain_to_enclave(); + b.drain_to_enclave(); + + // Whichever connection survived the race, it must actually carry traffic. + // Send again in both directions and require both messages to arrive. + const auto a_before = a.received; + const auto b_before = b.received; + + a.send_to(node_b_id, payload); + b.send_to(node_a_id, payload); + a.drain_to_host(); + b.drain_to_host(); + pump(); + a.drain_to_enclave(); + b.drain_to_enclave(); + + CHECK(b.received > b_before); + CHECK(a.received > a_before); + } + + net = nullptr; +} + +// Defending our own outgoing connection is only correct while a genuine race +// could be in progress. A peer which connects to us later is telling us it +// believes the link is broken, and it may have seen a failure we cannot see. +// If we refused it we would deny that peer the only means it has of repairing +// a connection that is dead in a way we cannot detect - reintroducing, in a +// different form, the very stall this change exists to prevent. +TEST_CASE("A later incoming connection is accepted, so a peer can repair") +{ + MockNet mock_net; + net = &mock_net; + + { + TestNode a("a", "10.0.0.1"); + TestNode b("b", "10.0.0.2"); + a.self_id = node_a_id; + b.self_id = node_b_id; + + // Node A has the lower ID, so it is the node which would otherwise defend + // its own outgoing connection and refuse B's. + REQUIRE(node_a_id < node_b_id); + + // Treat any existing outgoing connection as already settled, so B's + // connection is a repair attempt rather than part of a race. + a.connections->set_simultaneous_connect_window( + std::chrono::milliseconds(0)); + + a.learn_address(node_b_id, b.host, b.port); + b.learn_address(node_a_id, a.host, a.port); + a.drain_to_host(); + b.drain_to_host(); + + const auto payload = make_pre_vote(); + + // A dials B and the connection settles. + a.send_to(node_b_id, payload); + a.drain_to_host(); + complete_connects(); + pump(); + b.drain_to_enclave(); + REQUIRE(b.received > 0); + + // B observes that connection fail, but A does not - an asymmetric failure, + // which is the case that matters here. B drops its side; A is left holding + // an outgoing connection it still believes is fine, but whose far end is + // gone and whose writes now vanish. + mock_net.deliver_close = false; + auto* b_listener = mock_net.listeners.at(MockNet::key(b.host, b.port)); + auto* b_side = last_accepted_by(b_listener); + REQUIRE(b_side != nullptr); + b_side->behaviour->on_disconnect(); + + // B now has no connection to A, so sending makes it dial A. A must accept + // that connection, because it is B's only way to repair the link. + const auto a_before = a.received; + b.send_to(node_a_id, payload); + b.drain_to_host(); + complete_connects(); + pump(); + a.drain_to_enclave(); + + CHECK(a.received > a_before); + } + + net = nullptr; +} From 3db14109ee3a78f3ab3f17b50fb09b8a1f290a42 Mon Sep 17 00:00:00 2001 From: Amaury Chamayou Date: Fri, 28 Aug 2026 17:38:57 +0100 Subject: [PATCH 2/2] Add changelog entry for simultaneous connect fix Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 126fdff584c8..3304122c3e0b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,6 +25,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. ### Fixed +- Nodes which open node-to-node connections to each other at the same moment now agree on which of the two connections to keep, instead of each discarding the one the other is using. Previously both could be left holding a connection whose far end no longer existed, and if the resulting disconnection was not observed - for example because a network partition dropped it - the two nodes would silently stop exchanging consensus messages, stalling elections until one of them restarted (#8233). - Nodes from the previous service are now removed during disaster recovery instead of being retained as retired entries in `GET /node/network/nodes`, and `ledger_code.py` reports their code identities as removed (#8177). - Fixed an edge case where a follower could incorrectly commit to an abandoned fork while synchronising with the leader, causing it to become unavailable (#8172).