From 88a9caa46c1d692974652337b82e74033f5cf2d4 Mon Sep 17 00:00:00 2001 From: DcruBro Date: Sun, 26 Jul 2026 23:08:21 +0200 Subject: [PATCH] Reap dead outbound connection slots; strike disconnected peers from discovery. Two related connection-lifecycle fixes on top of the SIGPIPE fix. Reclaim outbound slots on peer disconnect: - TcpClient_ThreadProc fired on_disconnect but never cleared the outbound slot, closed the fd, joined the io thread, or freed the connection, so a dead peer permanently held its outboundClients[] slot. After MAX_CONS (32) churned connections the node could make no new outbound connections, and leaked fds/threads/memory. (The inbound side already self-reclaimed.) - Add a reaper (Node_ReapDeadOutbound) on the maintenance thread: under outboundLock it detaches dead (disconnect-notified) slots, then joins the io thread and destroys/frees each connection outside the lock. - Guard against use-after-free with a pin count on tcp_connection_t (TcpConnection_Pin/Unpin). The only cross-thread consumer holding a raw connection pointer across a blocking op is the `sync` command (via Node_GetBestOutboundPeer); it now pins the peer and unpins when done, and the reaper skips pinned connections. Discovery's snapshots run on the reaper's own thread, so they need no pin. - Node_GetBestOutboundPeer/GetClientList/GetPeerEndpoints skip disconnect-notified connections so a dead peer is never handed out. - Node_Destroy stops+joins the maintenance thread before tearing down outbound clients, so the reaper can't race shutdown. Strike disconnected peers from the discovery peer list: - Add NodeDiscovery_RemovePeer + Node_HandlePeerDisconnect: on disconnect, remove the peer from the known-peer table once no live connection (inbound or outbound) to its listen endpoint remains (Node_HasLiveConnectionTo; disconnect-notified conns don't count, so both directions dropping at once is handled). Wired into Node_Server_OnDisconnect and Node_Client_OnDisconnect. --- include/nets/nodediscovery.h | 4 ++ include/tcpd/tcpconnection.h | 11 ++++ src/main.c | 3 + src/nets/net_node.c | 116 ++++++++++++++++++++++++++++++----- src/nets/nodediscovery.c | 18 ++++++ src/tcpd/tcpconnection.c | 15 +++++ 6 files changed, 152 insertions(+), 15 deletions(-) diff --git a/include/nets/nodediscovery.h b/include/nets/nodediscovery.h index e03f41b..21bb314 100644 --- a/include/nets/nodediscovery.h +++ b/include/nets/nodediscovery.h @@ -23,6 +23,10 @@ void NodeDiscovery_OnGetPeers(node_discovery_t* disc, tcp_connection_t* fromConn // Decode a received PEERS payload and fold a couple of its endpoints into the known-peer table. void NodeDiscovery_OnPeersReceived(node_discovery_t* disc, tcp_connection_t* fromConn, const unsigned char* payload, size_t payloadLen); +// Strike a peer (by its listen endpoint) from the known-peer table. Called when a peer becomes +// logically disconnected (no remaining connection to it). +void NodeDiscovery_RemovePeer(node_discovery_t* disc, const struct sockaddr_storage* endpoint); + // Dump the known-peer table to stdout (for the CLI `peers` command). void NodeDiscovery_PrintPeers(node_discovery_t* disc); diff --git a/include/tcpd/tcpconnection.h b/include/tcpd/tcpconnection.h index 915ad5f..1e407c5 100644 --- a/include/tcpd/tcpconnection.h +++ b/include/tcpd/tcpconnection.h @@ -3,6 +3,7 @@ #include #include +#include #include #include #include @@ -37,6 +38,11 @@ struct tcp_connection_t { bool closing; bool disconnectedNotified; + // Non-zero while another thread holds a raw pointer to this connection taken from a + // lock-protected snapshot and used after releasing the lock. The reaper must not free a + // pinned connection. See TcpConnection_Pin/Unpin. + atomic_int pinCount; + unsigned char* dataBuf; size_t dataBufLen; size_t dataBufCap; @@ -75,4 +81,9 @@ void TcpConnection_RequestClose(tcp_connection_t* conn); void TcpConnection_MarkDisconnectNotified(tcp_connection_t* conn); bool TcpConnection_IsDisconnectNotified(tcp_connection_t* conn); +// Pin/unpin a connection so a background reaper won't free it while a caller still holds a raw +// pointer to it (e.g. across a blocking operation after releasing the collection lock). +void TcpConnection_Pin(tcp_connection_t* conn); +void TcpConnection_Unpin(tcp_connection_t* conn); + #endif diff --git a/src/main.c b/src/main.c index df622b5..7c235b1 100644 --- a/src/main.c +++ b/src/main.c @@ -1308,6 +1308,9 @@ int main(int argc, char* argv[]) { if ((uint64_t)Chain_Size(chain) >= peerHeight) break; continue; } + // Sync loop finished with this peer; release the pin taken by Node_GetBestOutboundPeer so + // the reaper may reclaim the slot if the peer has since disconnected. + TcpConnection_Unpin(peerConn); if (strcmp(cmd, "txpooldetail") == 0) { char* hashStr = strtok(NULL, " \t"); diff --git a/src/nets/net_node.c b/src/nets/net_node.c index 439d23f..fcfeeff 100644 --- a/src/nets/net_node.c +++ b/src/nets/net_node.c @@ -119,6 +119,47 @@ static int Node_HasOtherInboundFrom(net_node_t* node, const tcp_connection_t* se return found; } +// Returns non-zero if a connection OTHER than `exclude` to `endpoint` is still live. A connection +// that is itself mid-disconnect (disconnectedNotified) does not count as live - this is what lets +// us decide a peer is fully gone even when both its inbound and outbound drop simultaneously. +static int Node_HasLiveConnectionTo(net_node_t* node, const struct sockaddr_storage* endpoint, const tcp_connection_t* exclude) { + int found = 0; + + pthread_mutex_lock(&node->outboundLock); + for (size_t i = 0; i < MAX_CONS && !found; ++i) { + tcp_connection_t* c = node->outboundClients[i].connection; + if (!c || c == exclude || TcpConnection_IsDisconnectNotified(c)) continue; + struct sockaddr_storage ep; + if (Node_ConnListenEndpoint(c, &ep) && NetNode_EndpointEqual(&ep, endpoint)) found = 1; + } + pthread_mutex_unlock(&node->outboundLock); + if (found) return 1; + + if (node->server) { + pthread_mutex_lock(&node->server->clientsMutex); + for (size_t i = 0; i < node->server->maxClients && !found; ++i) { + tcp_connection_t* c = node->server->clientsArrPtr ? node->server->clientsArrPtr[i] : NULL; + if (!c || c == exclude || TcpConnection_IsDisconnectNotified(c)) continue; + struct sockaddr_storage ep; + if (Node_ConnListenEndpoint(c, &ep) && NetNode_EndpointEqual(&ep, endpoint)) found = 1; + } + pthread_mutex_unlock(&node->server->clientsMutex); + } + return found; +} + +// Called when a connection to a peer drops. Strikes the peer from the discovery table, but only +// once it is logically disconnected - i.e. no other live connection (inbound or outbound) to the +// same listen endpoint remains. Must be called from the disconnect callback while `conn` is still +// valid and outside outboundLock/clientsMutex. +static void Node_HandlePeerDisconnect(net_node_t* node, tcp_connection_t* conn) { + if (!node || !node->discovery || !conn) return; + struct sockaddr_storage ep; + if (!Node_ConnListenEndpoint(conn, &ep)) return; // never advertised an endpoint -> not tracked + if (Node_HasLiveConnectionTo(node, &ep, conn)) return; // still reachable via another connection + NodeDiscovery_RemovePeer(node->discovery, &ep); +} + int Node_ConnListenEndpoint(const tcp_connection_t* conn, struct sockaddr_storage* out) { if (!conn || !out) return 0; memset(out, 0, sizeof(*out)); @@ -170,7 +211,7 @@ size_t Node_GetPeerEndpoints(net_node_t* node, struct sockaddr_storage* outEndpo pthread_mutex_lock(&node->outboundLock); for (size_t i = 0; i < MAX_CONS && count < maxOut; ++i) { tcp_connection_t* c = node->outboundClients[i].connection; - if (!c) continue; + if (!c || TcpConnection_IsDisconnectNotified(c)) continue; // ignore connections that are tearing down struct sockaddr_storage ep; if (!Node_ConnListenEndpoint(c, &ep)) continue; int dup = 0; @@ -186,7 +227,7 @@ size_t Node_GetPeerEndpoints(net_node_t* node, struct sockaddr_storage* outEndpo pthread_mutex_lock(&node->server->clientsMutex); for (size_t i = 0; i < node->server->maxClients && count < maxOut; ++i) { tcp_connection_t* c = node->server->clientsArrPtr ? node->server->clientsArrPtr[i] : NULL; - if (!c) continue; + if (!c || TcpConnection_IsDisconnectNotified(c)) continue; // ignore connections that are tearing down struct sockaddr_storage ep; if (!Node_ConnListenEndpoint(c, &ep)) continue; int dup = 0; @@ -226,6 +267,41 @@ typedef enum { NODE_BLOCK_ACCEPTED = 2 } node_block_accept_result_t; +// Reclaims outbound slots whose peer has disconnected. Mirrors the inbound self-reclaim in +// TcpServer_clientthreadprocess: detach dead connections from their slots under outboundLock, then +// join their io threads and destroy/free them outside the lock. Pinned connections (a raw pointer +// is still held elsewhere, e.g. by an in-progress sync) are skipped and retried on a later tick. +static void Node_ReapDeadOutbound(net_node_t* node) { + if (!node) return; + + tcp_connection_t* dead[MAX_CONS]; + size_t deadCount = 0; + + pthread_mutex_lock(&node->outboundLock); + for (size_t i = 0; i < MAX_CONS; ++i) { + tcp_connection_t* c = node->outboundClients[i].connection; + if (!c) continue; + if (!TcpConnection_IsDisconnectNotified(c)) continue; // still live + if (atomic_load(&c->pinCount) != 0) continue; // someone holds a raw pointer; retry later + // Detach the dead connection from its slot and reset the slot to a clean free state. + node->outboundClients[i].connection = NULL; + node->outboundClients[i].peerBlockHeight = 0; + dead[deadCount++] = c; + } + pthread_mutex_unlock(&node->outboundLock); + + // Join + destroy outside the lock: the io thread's on_disconnect callback itself takes + // outboundLock, so joining under it would deadlock. + for (size_t i = 0; i < deadCount; ++i) { + tcp_connection_t* c = dead[i]; + if (!pthread_equal(c->ioThread, pthread_self())) { + pthread_join(c->ioThread, NULL); + } + TcpConnection_Destroy(c); + free(c); + } +} + static void* Node_MaintenanceThread(void* arg) { net_node_t* n = (net_node_t*)arg; if (!n) return NULL; @@ -238,6 +314,8 @@ static void* Node_MaintenanceThread(void* arg) { BalanceSheet_SaveToFile(chainDataDir); } } + // Reclaim outbound slots whose peer has disconnected so they can be reused. + Node_ReapDeadOutbound(n); // Peer discovery tick: ping/query connected peers and connect to the best-ping discoveries. if (n->discovery) { NodeDiscovery_Iterate(n->discovery); @@ -496,6 +574,14 @@ void Node_Destroy(net_node_t* node) { return; } + // Stop the maintenance thread first: it runs the outbound reaper (which touches outboundClients + // and outboundLock) and the discovery tick, so it must not run concurrently with the teardown + // below or against soon-to-be-destroyed state. + if (node->maintenanceRunning) { + node->maintenanceRunning = 0; + pthread_join(node->maintenanceThread, NULL); + } + for (size_t i = 0; i < MAX_CONS; ++i) { TcpClient_Destroy(&node->outboundClients[i]); } @@ -506,12 +592,6 @@ void Node_Destroy(net_node_t* node) { TcpServer_Destroy(node->server); } - // Stop maintenance thread (no more discovery ticks after this) - if (node->maintenanceRunning) { - node->maintenanceRunning = 0; - pthread_join(node->maintenanceThread, NULL); - } - // Tear down UDP + discovery. Stop UDP first so no pong/timeout callback races the destroy. if (node->udpNode) { UdpNode_Stop(node->udpNode); @@ -983,6 +1063,7 @@ void Node_Server_OnDisconnect(tcp_connection_t* client) { net_node_t* node = Node_FromConnection(client); Node_ForwardDisconnect(node, client); printf("Inbound node disconnected: %u\n", client ? client->connectionId : 0U); + Node_HandlePeerDisconnect(node, client); } void Node_Client_OnConnect(tcp_connection_t* client) { @@ -1202,6 +1283,7 @@ void Node_Client_OnDisconnect(tcp_connection_t* client) { Node_ForwardDisconnect(node, client); printf("Outbound node disconnected: %u\n", client ? client->connectionId : 0U); + Node_HandlePeerDisconnect(node, client); } int Node_GetBestOutboundPeer(net_node_t* node, tcp_connection_t** outConn, uint64_t* outHeight) { @@ -1212,13 +1294,16 @@ int Node_GetBestOutboundPeer(net_node_t* node, tcp_connection_t** outConn, uint6 pthread_mutex_lock(&node->outboundLock); for (size_t i = 0; i < MAX_CONS; ++i) { - if (node->outboundClients[i].connection) { - if (node->outboundClients[i].peerBlockHeight > bestH || best == NULL) { - best = node->outboundClients[i].connection; - bestH = node->outboundClients[i].peerBlockHeight; - } + tcp_connection_t* c = node->outboundClients[i].connection; + if (!c || TcpConnection_IsDisconnectNotified(c)) continue; // don't hand out a dead peer + if (best == NULL || node->outboundClients[i].peerBlockHeight > bestH) { + best = c; + bestH = node->outboundClients[i].peerBlockHeight; } } + // Pin the winner while still holding outboundLock so the reaper cannot free it out from under + // the caller (which uses the raw pointer after this lock is released). Caller must Unpin. + if (best) TcpConnection_Pin(best); pthread_mutex_unlock(&node->outboundLock); if (!best) return -1; @@ -1311,8 +1396,9 @@ void Node_GetClientList(net_node_t* node, tcp_connection_t** outClients, size_t* pthread_mutex_lock(&node->outboundLock); size_t count = 0; for (size_t i = 0; i < MAX_CONS; ++i) { - if (node->outboundClients[i].connection) { - outClients[count++] = node->outboundClients[i].connection; + tcp_connection_t* c = node->outboundClients[i].connection; + if (c && !TcpConnection_IsDisconnectNotified(c)) { // skip connections that are tearing down + outClients[count++] = c; } } pthread_mutex_unlock(&node->outboundLock); diff --git a/src/nets/nodediscovery.c b/src/nets/nodediscovery.c index 388097e..8e0c43d 100644 --- a/src/nets/nodediscovery.c +++ b/src/nets/nodediscovery.c @@ -274,6 +274,24 @@ void NodeDiscovery_OnPeersReceived(node_discovery_t* disc, tcp_connection_t* fro pthread_mutex_unlock(&disc->lock); } +void NodeDiscovery_RemovePeer(node_discovery_t* disc, const struct sockaddr_storage* endpoint) { + if (!disc || !endpoint) return; + pthread_mutex_lock(&disc->lock); + size_t n = DynArr_size(disc->peers); + for (size_t i = 0; i < n; ++i) { + discovered_peer_t* p = (discovered_peer_t*)DynArr_at(disc->peers, i); + if (Discovery_AddrEqual(&p->addr, endpoint)) { + char ip[INET6_ADDRSTRLEN] = {0}; + unsigned short port = 0; + Discovery_AddrToIpPort(&p->addr, ip, sizeof(ip), &port); + printf("NodeDiscovery: struck disconnected peer %s:%u from peer list\n", ip, port); + DynArr_remove(disc->peers, i); + break; + } + } + pthread_mutex_unlock(&disc->lock); +} + // ---- periodic tick ----------------------------------------------------------------------- void NodeDiscovery_Iterate(node_discovery_t* disc) { diff --git a/src/tcpd/tcpconnection.c b/src/tcpd/tcpconnection.c index 2ce2f75..24d4cfd 100644 --- a/src/tcpd/tcpconnection.c +++ b/src/tcpd/tcpconnection.c @@ -31,6 +31,7 @@ int TcpConnection_Init(tcp_connection_t* conn, int sockFd, const struct sockaddr conn->closing = false; conn->disconnectedNotified = false; + atomic_init(&conn->pinCount, 0); conn->dataBuf = NULL; conn->dataBufLen = 0; conn->dataBufCap = 0; @@ -262,6 +263,20 @@ bool TcpConnection_IsDisconnectNotified(tcp_connection_t* conn) { return notified; } +void TcpConnection_Pin(tcp_connection_t* conn) { + if (!conn) { + return; + } + atomic_fetch_add(&conn->pinCount, 1); +} + +void TcpConnection_Unpin(tcp_connection_t* conn) { + if (!conn) { + return; + } + atomic_fetch_sub(&conn->pinCount, 1); +} + static int extract_v4(const tcp_connection_t* conn, struct in_addr* v4out) { if (conn->addrFamily == AF_INET6) { const struct sockaddr_in6* a6 = (const struct sockaddr_in6*)&conn->peerAddr;