From 21fe73fb0175accb49f168a3f99fa4abc313560d Mon Sep 17 00:00:00 2001 From: DcruBro Date: Wed, 22 Jul 2026 23:18:57 +0200 Subject: [PATCH] Add NodeDiscovery: multi-hop peer crawl with UDP-ping preference - Implement NodeDiscovery engine: known-peer table (DynArr) with a per-tick seed/ping/query/connect state machine driven by the node maintenance thread; bounded multi-hop crawl (FANOUT peers per node, hop-capped) that connects to reachable peers lowest-ping-first - Add GET_PEERS/PEERS TCP opcodes for peer-list exchange, handled on both inbound and outbound connections - Measure UDP round-trip time and pass it to the on_pong callback (previously the send timestamp was only used for retries) - Advertise each node's listen port in HELLO/ACK_HELLO and store it per-connection, so inbound-only peers and non-default ports are discoverable (length-guarded parse; wire-compatible with old peers) - Wire a udp_node_t + node_discovery_t into net_node_t: init/start in Node_Create, tick in the maintenance loop, teardown in Node_Destroy (stop UDP before destroying discovery to avoid callback races) - Add Node_ConnListenEndpoint / Node_GetPeerEndpoints helpers to derive peers' listen endpoints (outbound: dialed port; inbound: advertised), with IPv4-mapped-IPv6 normalization and IP+port dedup - Match ping pong/timeout callbacks by peer address (the UDP layer owns the nonce), fixing discovered peers stuck UNREACHABLE - Ping the peer's listen port instead of the ephemeral TCP source port (fixes the original stub so pongs actually return) - Add `peers` CLI command to dump the discovery table (endpoint/hop/ state/ping) - Add discovery tunables to constants.h (fanout, max hops, target connections, timeouts, caps) --- include/constants.h | 12 + include/nets/net_node.h | 18 ++ include/nets/nodediscovery.h | 29 +++ include/packettype.h | 4 +- include/tcpd/tcpconnection.h | 4 + include/udpd/udpnode.h | 56 +++++ include/udpd/udppackettype.h | 14 ++ include/utils.h | 2 + src/main.c | 10 + src/nets/net_node.c | 244 +++++++++++++++++- src/nets/nodediscovery.c | 472 +++++++++++++++++++++++++++++++++++ src/udpd/udpnode.c | 379 ++++++++++++++++++++++++++++ 12 files changed, 1234 insertions(+), 10 deletions(-) create mode 100644 include/nets/nodediscovery.h create mode 100644 include/udpd/udpnode.h create mode 100644 include/udpd/udppackettype.h create mode 100644 src/nets/nodediscovery.c create mode 100644 src/udpd/udpnode.c diff --git a/include/constants.h b/include/constants.h index 43e16d3..248b343 100644 --- a/include/constants.h +++ b/include/constants.h @@ -13,6 +13,18 @@ #define MAX_CONS 32 // Some baseline for now #define LISTEN_PORT 9393 #define ECHO_PEERS 1 // If non-zero, automatically attempt to connect back to any inbound peers (helps form bidirectional peering) + +// Node discovery +#define DISCOVERY_FANOUT 2 // "A couple" - how many peers to query per round, and how many new peers to accept per PEERS response (keeps the crawl spread out) +#define DISCOVERY_MAX_HOPS 3 // How many hops away from us we keep crawling +#define DISCOVERY_TARGET_CONNECTIONS 8 // Desired outbound connection count discovery tries to reach (bounded by MAX_CONS) +#define DISCOVERY_MAX_KNOWN_PEERS 256 // Cap on the known-peer table size +#define DISCOVERY_PEERS_RESPONSE_CAP 8 // Max endpoints we put in a single PEERS response +#define DISCOVERY_MAX_PINGS_PER_TICK 8 // Cap on UDP pings sent per discovery tick +#define DISCOVERY_PING_TIMEOUT_MS 5000ULL // Backstop: a PINGED peer with no pong for this long is marked unreachable +#define DISCOVERY_PING_REFRESH_MS 60000ULL // Re-ping a reachable peer after this long to refresh its latency +#define DISCOVERY_QUERY_INTERVAL_MS 15000ULL // Minimum interval between GET_PEERS to the same peer +#define DISCOVERY_CONNECT_RETRY_MS 30000ULL // Minimum interval between connect attempts to the same discovered peer #define TCP_THREAD_STACK_SIZE (512 * 1024) // 512 KB. We could get away with like 128 KB since it's mostly just recv bufs, but it's good having some breathing room. // This is also for client threads. The server has the default (~8 MB on POSIX). diff --git a/include/nets/net_node.h b/include/nets/net_node.h index cf35826..22ff261 100644 --- a/include/nets/net_node.h +++ b/include/nets/net_node.h @@ -10,6 +10,10 @@ #include #include +#include + +// Forward declaration - the discovery state is defined in nodediscovery.c (opaque here). +typedef struct node_discovery node_discovery_t; #include @@ -40,6 +44,9 @@ typedef struct { pthread_t maintenanceThread; volatile int maintenanceRunning; int maintenanceIntervalMs; + // UDP ping/pong daemon (latency oracle) and peer discovery state + udp_node_t* udpNode; + node_discovery_t* discovery; } net_node_t; net_node_t* Node_Create(); @@ -71,4 +78,15 @@ void Node_Client_OnConnect(tcp_connection_t* client); void Node_Client_OnData(tcp_connection_t* client); void Node_Client_OnDisconnect(tcp_connection_t* client); +void Node_GetClientList(net_node_t* node, tcp_connection_t** outClients, size_t* outCount); + +// Computes a connection's peer listen endpoint (IP + advertised/dialed listen port) into *out. +// Outbound: the dialed peerAddr port already is the listen port. Inbound: uses peerListenPort. +// Returns non-zero on success (usable endpoint with a known, non-zero port), zero otherwise. +int Node_ConnListenEndpoint(const tcp_connection_t* conn, struct sockaddr_storage* out); + +// Fills outEndpoints with the listen endpoints of all current connections (inbound + outbound), +// deduped by IP+port. Returns the number of endpoints written (<= maxOut). +size_t Node_GetPeerEndpoints(net_node_t* node, struct sockaddr_storage* outEndpoints, size_t maxOut); + #endif diff --git a/include/nets/nodediscovery.h b/include/nets/nodediscovery.h new file mode 100644 index 0000000..e03f41b --- /dev/null +++ b/include/nets/nodediscovery.h @@ -0,0 +1,29 @@ +#ifndef NODEDISCOVERY_H +#define NODEDISCOVERY_H + +#include +#include + +// Create/destroy the peer-discovery state. Owns the known-peer table and its lock. +node_discovery_t* NodeDiscovery_Create(net_node_t* node, udp_node_t* udpNode); +void NodeDiscovery_Destroy(node_discovery_t* disc); + +// Periodic tick (driven by the node maintenance thread): seed currently-connected peers, +// UDP-ping newly-learned ones, query a couple of connected peers for more, and connect to +// the reachable peers with the lowest ping until we reach the target connection count. +void NodeDiscovery_Iterate(node_discovery_t* disc); + +// UDP latency callbacks (forwarded from the udp node via net_node thunks). +void NodeDiscovery_OnPong(node_discovery_t* disc, const struct sockaddr_storage* from, uint64_t nonce, uint64_t rttMs); +void NodeDiscovery_OnPingTimeout(node_discovery_t* disc, const struct sockaddr_storage* dest, uint64_t nonce); + +// TCP peer-exchange handlers (called from the net_node packet dispatch). +// Build a PEERS response (a sample of our peers, excluding the requester) and send it over fromConn. +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); + +// Dump the known-peer table to stdout (for the CLI `peers` command). +void NodeDiscovery_PrintPeers(node_discovery_t* disc); + +#endif diff --git a/include/packettype.h b/include/packettype.h index 44e0215..33a10b5 100644 --- a/include/packettype.h +++ b/include/packettype.h @@ -14,7 +14,9 @@ typedef enum { PACKET_TYPE_BROADCAST_TX = 7, // Here's a new transaction I want to share with the network PACKET_TYPE_ACK_TX = 8, // I have received your transaction, here's what I did with it (response to broadcast) PACKET_TYPE_ERROR = 9, // Something went wrong with the packet you sent me, here's an error message (can be response to any packet) - PACKET_TYPE_MAX = 10 + PACKET_TYPE_GET_PEERS = 10, // Who are your peers? Send me a few of them so I can discover more of the network + PACKET_TYPE_PEERS = 11, // Here are some of my peers' listen endpoints (response to GET_PEERS) + PACKET_TYPE_MAX = 12 } packet_type_t; static inline int PacketType_IsValid(uint8_t packetType) { diff --git a/include/tcpd/tcpconnection.h b/include/tcpd/tcpconnection.h index 98ade67..915ad5f 100644 --- a/include/tcpd/tcpconnection.h +++ b/include/tcpd/tcpconnection.h @@ -26,6 +26,10 @@ struct tcp_connection_t { uint32_t connectionId; tcp_connection_role_t role; + // Peer's advertised TCP/UDP listen port (learned from HELLO/ACK_HELLO). 0 until known. + // For OUTBOUND connections the peerAddr port already is the listen port; this matters for INBOUND peers. + uint16_t peerListenPort; + pthread_t ioThread; pthread_mutex_t sendLock; pthread_mutex_t stateLock; diff --git a/include/udpd/udpnode.h b/include/udpd/udpnode.h new file mode 100644 index 0000000..139ba74 --- /dev/null +++ b/include/udpd/udpnode.h @@ -0,0 +1,56 @@ +#ifndef UDP_NODE_H +#define UDP_NODE_H + +#include +#include +#include +#include + +#include + +#define UDP_LISTEN_PORT 9393 +#define UDP_PING_RETRY_INTERVAL_MS 1000 +#define UDP_PING_MAX_RETRIES 3 +#define UDP_MAX_PENDING_PINGS 64 + +typedef struct { + uint64_t nonce; + struct sockaddr_storage dest; + uint64_t lastSentMs; + int retries; + bool active; +} pending_ping_t; + +typedef struct udp_node { + int sockFd; // AF_INET6, IPV6_V6ONLY=1 + int sockFdV4; // AF_INET + + volatile int isRunning; + + pthread_t recvThreadV6; + pthread_t recvThreadV4; + pthread_t retryThread; + + pending_ping_t pendingPings[UDP_MAX_PENDING_PINGS]; + pthread_mutex_t pingsMutex; + + void (*on_pong)(struct udp_node* node, + const struct sockaddr_storage* from, + uint64_t nonce, int protoVersion, uint64_t rttMs, void* user); + void (*on_ping_timeout)(struct udp_node* node, + const struct sockaddr_storage* dest, + uint64_t nonce, void* user); + void* callbackUser; +} udp_node_t; + +int UdpNode_Init(udp_node_t* node, uint16_t port); +void UdpNode_SetCallbacks(udp_node_t* node, + void (*on_pong)(udp_node_t*, const struct sockaddr_storage*, uint64_t, int, uint64_t, void*), + void (*on_ping_timeout)(udp_node_t*, const struct sockaddr_storage*, uint64_t, void*), + void* user); +int UdpNode_Start(udp_node_t* node); +void UdpNode_Stop(udp_node_t* node); +void UdpNode_Destroy(udp_node_t* node); +int UdpNode_SendPing(udp_node_t* node, const struct sockaddr_storage* dest); + +#endif diff --git a/include/udpd/udppackettype.h b/include/udpd/udppackettype.h new file mode 100644 index 0000000..a8f598d --- /dev/null +++ b/include/udpd/udppackettype.h @@ -0,0 +1,14 @@ +#ifndef UDP_PACKET_TYPE_H +#define UDP_PACKET_TYPE_H + +typedef enum { + UDP_PACKET_TYPE_NONE = 0, + UDP_PACKET_TYPE_PING = 1, + UDP_PACKET_TYPE_PONG = 2, +} udp_packet_type_t; + +// Wire sizes in bytes +#define UDP_PING_WIRE_SIZE 9 // 1 (type) + 8 (nonce) +#define UDP_PONG_WIRE_SIZE 13 // 1 (type) + 8 (nonce) + 4 (proto_version) + +#endif diff --git a/include/utils.h b/include/utils.h index 68d7d18..35747ee 100644 --- a/include/utils.h +++ b/include/utils.h @@ -14,6 +14,8 @@ typedef struct { uint8_t bytes[32]; } key32_t; +#define PROTO_VERSION 1 + static inline uint32_t hash_key32(key32_t k) { uint32_t hash = 2166136261u; for (int i = 0; i < 32; i++) { diff --git a/src/main.c b/src/main.c index 92fab1b..5de6a27 100644 --- a/src/main.c +++ b/src/main.c @@ -19,6 +19,7 @@ #include #include +#include #include #include @@ -1464,6 +1465,15 @@ int main(int argc, char* argv[]) { continue; } + if (strcmp(cmd, "peers") == 0) { + if (strtok(NULL, " \t")) { + printf("usage: peers\n"); + continue; + } + NodeDiscovery_PrintPeers(node->discovery); + continue; + } + if (strcmp(cmd, "flushchain") == 0) { if (FlushChainAndSheet(chain, chainDataDir, currentSupply, currentReward)) { printf("chain flushed\n"); diff --git a/src/nets/net_node.c b/src/nets/net_node.c index bad88cb..b906d1b 100644 --- a/src/nets/net_node.c +++ b/src/nets/net_node.c @@ -1,4 +1,5 @@ #include +#include #include #include @@ -12,6 +13,8 @@ #include #include #include +#include +#include static net_node_t* Node_FromConnection(tcp_connection_t* conn) { if (!conn) { @@ -29,6 +32,125 @@ static uint64_t Node_GetCurrentBlockHeight(void) { return currentBlockHeight; } +// Compares two listen endpoints (family + IP + port). +static int NetNode_EndpointEqual(const struct sockaddr_storage* a, const struct sockaddr_storage* b) { + if (a->ss_family != b->ss_family) return 0; + if (a->ss_family == AF_INET) { + const struct sockaddr_in* x = (const struct sockaddr_in*)a; + const struct sockaddr_in* y = (const struct sockaddr_in*)b; + return x->sin_port == y->sin_port && + memcmp(&x->sin_addr, &y->sin_addr, sizeof(struct in_addr)) == 0; + } + if (a->ss_family == AF_INET6) { + const struct sockaddr_in6* x = (const struct sockaddr_in6*)a; + const struct sockaddr_in6* y = (const struct sockaddr_in6*)b; + return x->sin6_port == y->sin6_port && + memcmp(&x->sin6_addr, &y->sin6_addr, sizeof(struct in6_addr)) == 0; + } + return 0; +} + +int Node_ConnListenEndpoint(const tcp_connection_t* conn, struct sockaddr_storage* out) { + if (!conn || !out) return 0; + memset(out, 0, sizeof(*out)); + + // Determine the peer's listen port. For an outbound connection the port we dialed already + // is the peer's listen port; for an inbound one it is the port advertised in HELLO. + unsigned short listenP; + if (conn->role == TCP_CONNECTION_ROLE_OUTBOUND) { + listenP = (conn->addrFamily == AF_INET6) + ? ntohs(((const struct sockaddr_in6*)&conn->peerAddr)->sin6_port) + : ntohs(((const struct sockaddr_in*)&conn->peerAddr)->sin_port); + } else { + listenP = conn->peerListenPort; + } + if (listenP == 0) return 0; // unknown listen port -> not a usable endpoint + + if (conn->addrFamily == AF_INET) { + const struct sockaddr_in* a = (const struct sockaddr_in*)&conn->peerAddr; + struct sockaddr_in* o = (struct sockaddr_in*)out; + o->sin_family = AF_INET; + o->sin_addr = a->sin_addr; + o->sin_port = htons(listenP); + return 1; + } + if (conn->addrFamily == AF_INET6) { + const struct sockaddr_in6* a = (const struct sockaddr_in6*)&conn->peerAddr; + if (IN6_IS_ADDR_V4MAPPED(&a->sin6_addr)) { + // Normalise IPv4-mapped IPv6 to plain IPv4. + struct sockaddr_in* o = (struct sockaddr_in*)out; + o->sin_family = AF_INET; + memcpy(&o->sin_addr, ((const uint8_t*)&a->sin6_addr) + 12, sizeof(struct in_addr)); + o->sin_port = htons(listenP); + } else { + struct sockaddr_in6* o = (struct sockaddr_in6*)out; + o->sin6_family = AF_INET6; + o->sin6_addr = a->sin6_addr; + o->sin6_port = htons(listenP); + } + return 1; + } + return 0; +} + +size_t Node_GetPeerEndpoints(net_node_t* node, struct sockaddr_storage* outEndpoints, size_t maxOut) { + if (!node || !outEndpoints || maxOut == 0) return 0; + size_t count = 0; + + // Outbound connections + 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; + struct sockaddr_storage ep; + if (!Node_ConnListenEndpoint(c, &ep)) continue; + int dup = 0; + for (size_t k = 0; k < count; ++k) { + if (NetNode_EndpointEqual(&outEndpoints[k], &ep)) { dup = 1; break; } + } + if (!dup) outEndpoints[count++] = ep; + } + pthread_mutex_unlock(&node->outboundLock); + + // Inbound connections + if (node->server) { + 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; + struct sockaddr_storage ep; + if (!Node_ConnListenEndpoint(c, &ep)) continue; + int dup = 0; + for (size_t k = 0; k < count; ++k) { + if (NetNode_EndpointEqual(&outEndpoints[k], &ep)) { dup = 1; break; } + } + if (!dup) outEndpoints[count++] = ep; + } + pthread_mutex_unlock(&node->server->clientsMutex); + } + + return count; +} + +// Thunks routing UDP ping/pong events into the discovery state. +static void Node_OnPongThunk(udp_node_t* udp, const struct sockaddr_storage* from, + uint64_t nonce, int protoVersion, uint64_t rttMs, void* user) { + (void)udp; (void)protoVersion; + net_node_t* node = (net_node_t*)user; + if (node && node->discovery) { + NodeDiscovery_OnPong(node->discovery, from, nonce, rttMs); + } +} + +static void Node_OnPingTimeoutThunk(udp_node_t* udp, const struct sockaddr_storage* dest, + uint64_t nonce, void* user) { + (void)udp; + net_node_t* node = (net_node_t*)user; + if (node && node->discovery) { + NodeDiscovery_OnPingTimeout(node->discovery, dest, nonce); + } +} + typedef enum { NODE_BLOCK_REJECTED = 0, NODE_BLOCK_ORPHAN_QUEUED = 1, @@ -47,6 +169,10 @@ static void* Node_MaintenanceThread(void* arg) { BalanceSheet_SaveToFile(chainDataDir); } } + // Peer discovery tick: ping/query connected peers and connect to the best-ping discoveries. + if (n->discovery) { + NodeDiscovery_Iterate(n->discovery); + } sleep_for_milliseconds((uint64_t)n->maintenanceIntervalMs); } return NULL; @@ -266,6 +392,25 @@ net_node_t* Node_Create() { OrphanPool_Init(); + // Start the UDP ping/pong daemon (latency oracle) and peer discovery. Non-fatal on failure; + // the node still works without discovery, it just won't crawl for new peers. + node->udpNode = (udp_node_t*)malloc(sizeof(udp_node_t)); + if (node->udpNode) { + if (UdpNode_Init(node->udpNode, (uint16_t)listenPort) == 0) { + UdpNode_SetCallbacks(node->udpNode, Node_OnPongThunk, Node_OnPingTimeoutThunk, node); + if (UdpNode_Start(node->udpNode) == 0) { + node->discovery = NodeDiscovery_Create(node, node->udpNode); + } else { + UdpNode_Destroy(node->udpNode); + free(node->udpNode); + node->udpNode = NULL; + } + } else { + free(node->udpNode); + node->udpNode = NULL; + } + } + // Start maintenance thread node->maintenanceRunning = 1; node->maintenanceIntervalMs = 1000; // 1s @@ -292,12 +437,26 @@ void Node_Destroy(net_node_t* node) { TcpServer_Destroy(node->server); } - // Stop maintenance thread + // 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); + } + if (node->discovery) { + NodeDiscovery_Destroy(node->discovery); + node->discovery = NULL; + } + if (node->udpNode) { + UdpNode_Destroy(node->udpNode); + free(node->udpNode); + node->udpNode = NULL; + } + OrphanPool_Destroy(); TxMempool_Destroy(); @@ -482,11 +641,20 @@ void Node_Server_OnData(tcp_connection_t* client) { memcpy(&protoVersion, payload, sizeof(protoVersion)); memcpy(&blockHeight, payload + sizeof(protoVersion), sizeof(blockHeight)); - // TODO: Save these somewhere and maybe respond - printf("Received HELLO from node %u: protoVersion=%u, blockHeight=%" PRIu64 "\n", - client ? client->connectionId : 0U, protoVersion, blockHeight); + // Optional trailing listen port. This inbound peer's source port is ephemeral, + // so we record the port it actually listens on to make it discoverable/reachable. + // Length-guarded so older peers that omit it still work. + if (client && payloadLen >= sizeof(uint32_t) + sizeof(uint64_t) + sizeof(uint16_t)) { + uint16_t peerListenPort; + memcpy(&peerListenPort, payload + sizeof(protoVersion) + sizeof(blockHeight), sizeof(peerListenPort)); + client->peerListenPort = peerListenPort; + } - // Craft and send ACK_HELLO + printf("Received HELLO from node %u: protoVersion=%u, blockHeight=%" PRIu64 ", listenPort=%u\n", + client ? client->connectionId : 0U, protoVersion, blockHeight, + client ? client->peerListenPort : 0U); + + // Craft and send ACK_HELLO (echo protoVersion, our height, and our own listen port) uint8_t ackBuf[100]; uint8_t* ackData = ackBuf; size_t ackOffset = 0; @@ -495,6 +663,9 @@ void Node_Server_OnData(tcp_connection_t* client) { uint64_t currentHeight = Node_GetCurrentBlockHeight(); memcpy(ackData + ackOffset, ¤tHeight, sizeof(currentHeight)); ackOffset += sizeof(currentHeight); + uint16_t myListenPort = (uint16_t)listenPort; + memcpy(ackData + ackOffset, &myListenPort, sizeof(myListenPort)); + ackOffset += sizeof(myListenPort); Node_SendPacket(Node_FromConnection(client), client, PACKET_TYPE_ACK_HELLO, ackData, ackOffset); @@ -690,12 +861,26 @@ void Node_Server_OnData(tcp_connection_t* client) { printf("Received packet type %u from node %u with message: %s\n", (unsigned int)packetType, client ? client->connectionId : 0U, text); free(text); - + + break; + } + case PACKET_TYPE_GET_PEERS: { + net_node_t* dnode = Node_FromConnection(client); + if (dnode && dnode->discovery) { + NodeDiscovery_OnGetPeers(dnode->discovery, client); + } + break; + } + case PACKET_TYPE_PEERS: { + net_node_t* dnode = Node_FromConnection(client); + if (dnode && dnode->discovery) { + NodeDiscovery_OnPeersReceived(dnode->discovery, client, payload, payloadLen); + } break; } default: return; - } + } net_node_t* node = Node_FromConnection(client); Node_ForwardData(node, client, payload, payloadLen); @@ -718,12 +903,16 @@ void Node_Client_OnConnect(tcp_connection_t* client) { uint8_t* data = buf; size_t offset = 0; - uint32_t protoVersion = 1; // little-endian + uint32_t protoVersion = PROTO_VERSION; // little-endian uint64_t blockHeight = Node_GetCurrentBlockHeight(); memcpy((unsigned char*)data + offset, &protoVersion, sizeof(protoVersion)); // This is technically "unsafe", but I honestly just don't give a shit at this point offset += sizeof(protoVersion); memcpy((unsigned char*)data + offset, &blockHeight, sizeof(blockHeight)); offset += sizeof(blockHeight); + // Advertise the port we listen on so the peer can share us with others (and reach us back) + uint16_t myListenPort = (uint16_t)listenPort; + memcpy((unsigned char*)data + offset, &myListenPort, sizeof(myListenPort)); + offset += sizeof(myListenPort); Node_SendPacket(node, client, PACKET_TYPE_HELLO, data, offset); } @@ -760,6 +949,14 @@ void Node_Client_OnData(tcp_connection_t* client) { memcpy(&protoVersion, payload, sizeof(protoVersion)); memcpy(&blockHeight, payload + sizeof(protoVersion), sizeof(blockHeight)); + // Optional trailing listen port (for outbound peers the dialed port is already the + // listen port, but record the advertised one for consistency). Length-guarded. + if (client && payloadLen >= sizeof(uint32_t) + sizeof(uint64_t) + sizeof(uint16_t)) { + uint16_t peerListenPort; + memcpy(&peerListenPort, payload + sizeof(protoVersion) + sizeof(blockHeight), sizeof(peerListenPort)); + client->peerListenPort = peerListenPort; + } + printf("Received ACK_HELLO from node %u with protoVersion %u and blockHeight %" PRIu64 "\n", client ? client->connectionId : 0U, protoVersion, blockHeight); // Store peer-advertised height on matching outbound client @@ -867,7 +1064,21 @@ void Node_Client_OnData(tcp_connection_t* client) { printf("Received packet type %u from node %u with message: %s\n", (unsigned int)packetType, client ? client->connectionId : 0U, text); free(text); - + + break; + } + case PACKET_TYPE_GET_PEERS: { + net_node_t* dnode = Node_FromConnection(client); + if (dnode && dnode->discovery) { + NodeDiscovery_OnGetPeers(dnode->discovery, client); + } + break; + } + case PACKET_TYPE_PEERS: { + net_node_t* dnode = Node_FromConnection(client); + if (dnode && dnode->discovery) { + NodeDiscovery_OnPeersReceived(dnode->discovery, client, payload, payloadLen); + } break; } default: @@ -1000,3 +1211,18 @@ void Node_BroadcastChainRange(net_node_t* node, size_t startHeightInclusive, tcp Block_Destroy(blk); } } + +void Node_GetClientList(net_node_t* node, tcp_connection_t** outClients, size_t* outCount) { + if (!node || !outClients || !outCount) return; + + 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; + } + } + pthread_mutex_unlock(&node->outboundLock); + + *outCount = count; +} diff --git a/src/nets/nodediscovery.c b/src/nets/nodediscovery.c new file mode 100644 index 0000000..388097e --- /dev/null +++ b/src/nets/nodediscovery.c @@ -0,0 +1,472 @@ +#include + +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include + +// Wire layout of a single peer endpoint inside a PEERS payload: +// [uint8 family (4 or 6)][uint8 ip[16]][uint16 port (host order)] -> 19 bytes +// (Host-endian raw layout, consistent with the rest of the protocol.) +#define DISCOVERY_WIRE_ENTRY_SIZE (1 + 16 + 2) + +typedef enum { + DISCOVERY_STATE_NEW = 0, // learned, not yet pinged + DISCOVERY_STATE_PINGED, // first ping in flight, reachability unknown + DISCOVERY_STATE_REACHABLE, // pong received, latency known + DISCOVERY_STATE_CONNECTED, // currently a live outbound connection + DISCOVERY_STATE_UNREACHABLE // ping timed out +} discovery_state_t; + +typedef struct { + struct sockaddr_storage addr; // listen endpoint (port already set to the peer's listen port) + uint64_t pingMs; // measured UDP RTT, UINT64_MAX if unknown + uint32_t hop; // distance from us (0 = directly connected) + discovery_state_t state; + int pingPending; // 1 while a ping is outstanding (matched by address on pong/timeout). + // The UDP layer generates its own nonce, so we can't match by nonce here. + uint64_t lastPingMs; // when we last sent a ping + uint64_t lastQueryMs; // when we last sent GET_PEERS to it + uint64_t lastConnectMs; // when we last attempted a connect to it +} discovered_peer_t; + +struct node_discovery { + net_node_t* node; + udp_node_t* udpNode; + DynArr* peers; // of discovered_peer_t + pthread_mutex_t lock; +}; + +// ---- small helpers (most assume the caller holds disc->lock) ------------------------------ + +static int Discovery_AddrEqual(const struct sockaddr_storage* a, const struct sockaddr_storage* b) { + if (a->ss_family != b->ss_family) return 0; + if (a->ss_family == AF_INET) { + const struct sockaddr_in* x = (const struct sockaddr_in*)a; + const struct sockaddr_in* y = (const struct sockaddr_in*)b; + return x->sin_port == y->sin_port && + memcmp(&x->sin_addr, &y->sin_addr, sizeof(struct in_addr)) == 0; + } + if (a->ss_family == AF_INET6) { + const struct sockaddr_in6* x = (const struct sockaddr_in6*)a; + const struct sockaddr_in6* y = (const struct sockaddr_in6*)b; + return x->sin6_port == y->sin6_port && + memcmp(&x->sin6_addr, &y->sin6_addr, sizeof(struct in6_addr)) == 0; + } + return 0; +} + +static discovered_peer_t* Discovery_FindPtr(node_discovery_t* disc, const struct sockaddr_storage* addr) { + 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, addr)) return p; + } + return NULL; +} + +// Insert addr if not already present. Returns a pointer to the (existing or new) entry, or NULL +// if the table is full. Note: the returned pointer is invalidated by any later push_back. +static discovered_peer_t* Discovery_Upsert(node_discovery_t* disc, const struct sockaddr_storage* addr, uint32_t hop) { + discovered_peer_t* existing = Discovery_FindPtr(disc, addr); + if (existing) { + if (hop < existing->hop) existing->hop = hop; // keep the shortest known distance + return existing; + } + if (DynArr_size(disc->peers) >= DISCOVERY_MAX_KNOWN_PEERS) return NULL; + + discovered_peer_t np; + memset(&np, 0, sizeof(np)); + np.addr = *addr; + np.pingMs = UINT64_MAX; + np.hop = hop; + np.state = DISCOVERY_STATE_NEW; + DynArr_push_back(disc->peers, &np); + return (discovered_peer_t*)DynArr_at(disc->peers, DynArr_size(disc->peers) - 1); +} + +static int Discovery_AddrToWire(const struct sockaddr_storage* addr, unsigned char out[DISCOVERY_WIRE_ENTRY_SIZE]) { + memset(out, 0, DISCOVERY_WIRE_ENTRY_SIZE); + if (addr->ss_family == AF_INET) { + const struct sockaddr_in* a = (const struct sockaddr_in*)addr; + out[0] = 4; + memcpy(out + 1, &a->sin_addr, sizeof(struct in_addr)); + uint16_t port = ntohs(a->sin_port); + memcpy(out + 1 + 16, &port, sizeof(port)); + return 1; + } + if (addr->ss_family == AF_INET6) { + const struct sockaddr_in6* a = (const struct sockaddr_in6*)addr; + out[0] = 6; + memcpy(out + 1, &a->sin6_addr, sizeof(struct in6_addr)); + uint16_t port = ntohs(a->sin6_port); + memcpy(out + 1 + 16, &port, sizeof(port)); + return 1; + } + return 0; +} + +static int Discovery_WireToAddr(const unsigned char in[DISCOVERY_WIRE_ENTRY_SIZE], struct sockaddr_storage* out) { + memset(out, 0, sizeof(*out)); + uint8_t fam = in[0]; + uint16_t port; + memcpy(&port, in + 1 + 16, sizeof(port)); + if (fam == 4) { + struct sockaddr_in* a = (struct sockaddr_in*)out; + a->sin_family = AF_INET; + memcpy(&a->sin_addr, in + 1, sizeof(struct in_addr)); + a->sin_port = htons(port); + return port != 0; + } + if (fam == 6) { + struct sockaddr_in6* a = (struct sockaddr_in6*)out; + a->sin6_family = AF_INET6; + memcpy(&a->sin6_addr, in + 1, sizeof(struct in6_addr)); + a->sin6_port = htons(port); + return port != 0; + } + return 0; +} + +static int Discovery_AddrToIpPort(const struct sockaddr_storage* addr, char* ipOut, size_t ipLen, unsigned short* portOut) { + if (addr->ss_family == AF_INET) { + const struct sockaddr_in* a = (const struct sockaddr_in*)addr; + if (!inet_ntop(AF_INET, &a->sin_addr, ipOut, (socklen_t)ipLen)) return 0; + *portOut = ntohs(a->sin_port); + return 1; + } + if (addr->ss_family == AF_INET6) { + const struct sockaddr_in6* a = (const struct sockaddr_in6*)addr; + if (!inet_ntop(AF_INET6, &a->sin6_addr, ipOut, (socklen_t)ipLen)) return 0; + *portOut = ntohs(a->sin6_port); + return 1; + } + return 0; +} + +// ---- lifecycle --------------------------------------------------------------------------- + +node_discovery_t* NodeDiscovery_Create(net_node_t* node, udp_node_t* udpNode) { + if (!node || !udpNode) return NULL; + node_discovery_t* disc = (node_discovery_t*)malloc(sizeof(node_discovery_t)); + if (!disc) return NULL; + memset(disc, 0, sizeof(*disc)); + disc->node = node; + disc->udpNode = udpNode; + disc->peers = DYNARR_CREATE(discovered_peer_t, 16); + if (!disc->peers) { + free(disc); + return NULL; + } + pthread_mutex_init(&disc->lock, NULL); + return disc; +} + +void NodeDiscovery_Destroy(node_discovery_t* disc) { + if (!disc) return; + if (disc->peers) DynArr_destroy(disc->peers); + pthread_mutex_destroy(&disc->lock); + free(disc); +} + +// ---- UDP latency callbacks --------------------------------------------------------------- + +void NodeDiscovery_OnPong(node_discovery_t* disc, const struct sockaddr_storage* from, uint64_t nonce, uint64_t rttMs) { + if (!disc || !from) return; + (void)nonce; // UDP layer owns the nonce; we match the peer by its reply address instead. + pthread_mutex_lock(&disc->lock); + discovered_peer_t* p = Discovery_FindPtr(disc, from); + if (p && p->pingPending) { + p->pingMs = rttMs; + p->pingPending = 0; + if (p->state == DISCOVERY_STATE_PINGED) p->state = DISCOVERY_STATE_REACHABLE; + } + pthread_mutex_unlock(&disc->lock); +} + +void NodeDiscovery_OnPingTimeout(node_discovery_t* disc, const struct sockaddr_storage* dest, uint64_t nonce) { + if (!disc || !dest) return; + (void)nonce; // matched by the destination address we pinged + pthread_mutex_lock(&disc->lock); + discovered_peer_t* p = Discovery_FindPtr(disc, dest); + if (p && p->pingPending) { + p->pingPending = 0; + // Only demote a peer whose reachability was still unknown; a refresh ping that times + // out on an already-connected/reachable peer must not drop it. + if (p->state == DISCOVERY_STATE_PINGED) p->state = DISCOVERY_STATE_UNREACHABLE; + } + pthread_mutex_unlock(&disc->lock); +} + +// ---- TCP peer exchange ------------------------------------------------------------------- + +void NodeDiscovery_OnGetPeers(node_discovery_t* disc, tcp_connection_t* fromConn) { + if (!disc || !fromConn) return; + + // Snapshot our current peers' listen endpoints (inbound + outbound). + struct sockaddr_storage all[MAX_CONS * 2]; + size_t total = Node_GetPeerEndpoints(disc->node, all, sizeof(all) / sizeof(all[0])); + + struct sockaddr_storage reqEndpoint; + int haveReq = Node_ConnListenEndpoint(fromConn, &reqEndpoint); + + // Build the response payload: [uint16 count][entries...], capped and sampled for spread. + unsigned char payload[sizeof(uint16_t) + DISCOVERY_PEERS_RESPONSE_CAP * DISCOVERY_WIRE_ENTRY_SIZE]; + size_t offset = sizeof(uint16_t); + uint16_t count = 0; + + size_t startIdx = total ? (size_t)(random_four_byte() % total) : 0; + for (size_t k = 0; k < total && count < DISCOVERY_PEERS_RESPONSE_CAP; ++k) { + size_t idx = (startIdx + k) % total; + if (haveReq && Discovery_AddrEqual(&all[idx], &reqEndpoint)) continue; // don't tell them about themselves + unsigned char entry[DISCOVERY_WIRE_ENTRY_SIZE]; + if (!Discovery_AddrToWire(&all[idx], entry)) continue; + memcpy(payload + offset, entry, DISCOVERY_WIRE_ENTRY_SIZE); + offset += DISCOVERY_WIRE_ENTRY_SIZE; + count++; + } + memcpy(payload, &count, sizeof(count)); + + Node_SendPacket(disc->node, fromConn, PACKET_TYPE_PEERS, payload, offset); +} + +void NodeDiscovery_OnPeersReceived(node_discovery_t* disc, tcp_connection_t* fromConn, const unsigned char* payload, size_t payloadLen) { + if (!disc || !payload || payloadLen < sizeof(uint16_t)) return; + + uint16_t count; + memcpy(&count, payload, sizeof(count)); + size_t need = sizeof(uint16_t) + (size_t)count * DISCOVERY_WIRE_ENTRY_SIZE; + if (payloadLen < need) return; // malformed / truncated + + // Determine the hop distance of the peer that answered, so its peers land one hop further out. + struct sockaddr_storage srcEndpoint; + int haveSrc = fromConn ? Node_ConnListenEndpoint(fromConn, &srcEndpoint) : 0; + + pthread_mutex_lock(&disc->lock); + + uint32_t srcHop = 0; + if (haveSrc) { + discovered_peer_t* srcp = Discovery_FindPtr(disc, &srcEndpoint); + if (srcp) srcHop = srcp->hop; + } + uint32_t newHop = srcHop + 1; + + // Fold in at most DISCOVERY_FANOUT *new* endpoints (a couple per node -> keeps the crawl spread). + if (newHop <= DISCOVERY_MAX_HOPS) { + int added = 0; + for (uint16_t i = 0; i < count && added < DISCOVERY_FANOUT; ++i) { + const unsigned char* entry = payload + sizeof(uint16_t) + (size_t)i * DISCOVERY_WIRE_ENTRY_SIZE; + struct sockaddr_storage ep; + if (!Discovery_WireToAddr(entry, &ep)) continue; + if (Discovery_FindPtr(disc, &ep) != NULL) continue; // already known -> doesn't count toward fanout + if (Discovery_Upsert(disc, &ep, newHop) != NULL) added++; + } + } + + pthread_mutex_unlock(&disc->lock); +} + +// ---- periodic tick ----------------------------------------------------------------------- + +void NodeDiscovery_Iterate(node_discovery_t* disc) { + if (!disc || !disc->node || !disc->udpNode) return; + uint64_t now = get_current_time_ms(); + + // Snapshot current outbound connections and their listen endpoints (used for querying and + // for the "already connected?" checks below). + tcp_connection_t* outConns[MAX_CONS]; + size_t outCount = 0; + Node_GetClientList(disc->node, outConns, &outCount); + + struct sockaddr_storage outEndpoints[MAX_CONS]; + size_t outEpCount = 0; + for (size_t i = 0; i < outCount; ++i) { + struct sockaddr_storage ep; + if (Node_ConnListenEndpoint(outConns[i], &ep)) { + outEndpoints[outEpCount++] = ep; + } + } + + // Deferred network actions, collected under the lock and executed after releasing it + // (Node_ConnectPeer / Node_SendPacket must not run while holding disc->lock). + tcp_connection_t* toQuery[MAX_CONS]; + size_t toQueryCount = 0; + struct { char ip[INET6_ADDRSTRLEN]; unsigned short port; } toConnect[MAX_CONS]; + size_t toConnectCount = 0; + + pthread_mutex_lock(&disc->lock); + + // 1. Seed: upsert connected (outbound) peers as CONNECTED at hop 0. + for (size_t i = 0; i < outEpCount; ++i) { + discovered_peer_t* p = Discovery_Upsert(disc, &outEndpoints[i], 0); + if (p) { + p->hop = 0; + p->state = DISCOVERY_STATE_CONNECTED; + } + } + // Demote entries still marked CONNECTED that are no longer in the outbound set. + { + 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 (p->state != DISCOVERY_STATE_CONNECTED) continue; + int stillConnected = 0; + for (size_t j = 0; j < outEpCount; ++j) { + if (Discovery_AddrEqual(&p->addr, &outEndpoints[j])) { stillConnected = 1; break; } + } + if (!stillConnected) { + p->state = (p->pingMs != UINT64_MAX) ? DISCOVERY_STATE_REACHABLE : DISCOVERY_STATE_NEW; + } + } + } + + // 2. Ping NEW peers (and refresh stale REACHABLE ones), capped per tick. + { + int pings = 0; + size_t n = DynArr_size(disc->peers); + for (size_t i = 0; i < n && pings < DISCOVERY_MAX_PINGS_PER_TICK; ++i) { + discovered_peer_t* p = (discovered_peer_t*)DynArr_at(disc->peers, i); + int shouldPing = 0; + if (!p->pingPending) { + if (p->state == DISCOVERY_STATE_NEW) { + shouldPing = 1; + } else if (p->state == DISCOVERY_STATE_REACHABLE && + (now - p->lastPingMs) > DISCOVERY_PING_REFRESH_MS) { + shouldPing = 1; + } + } + if (!shouldPing) continue; + + p->pingPending = 1; + p->lastPingMs = now; + if (p->state == DISCOVERY_STATE_NEW) p->state = DISCOVERY_STATE_PINGED; + UdpNode_SendPing(disc->udpNode, &p->addr); + pings++; + } + } + + // 3. Timeout backstop (in case the UDP layer's own timeout callback is missed). + { + 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 (p->state == DISCOVERY_STATE_PINGED && p->pingPending && + (now - p->lastPingMs) > DISCOVERY_PING_TIMEOUT_MS) { + p->pingPending = 0; + p->state = DISCOVERY_STATE_UNREACHABLE; + } + } + } + + // 4. Query up to DISCOVERY_FANOUT connected peers with GET_PEERS, preferring the lowest ping + // and skipping ones we queried recently or that are already at the hop horizon. + { + int queries = 0; + while (queries < DISCOVERY_FANOUT) { + size_t bestIdx = outCount; // sentinel = none + uint64_t bestPing = UINT64_MAX; + for (size_t i = 0; i < outCount; ++i) { + struct sockaddr_storage ep; + if (!Node_ConnListenEndpoint(outConns[i], &ep)) continue; + discovered_peer_t* p = Discovery_FindPtr(disc, &ep); + if (!p) continue; + if (p->hop >= DISCOVERY_MAX_HOPS) continue; + if (p->lastQueryMs != 0 && (now - p->lastQueryMs) < DISCOVERY_QUERY_INTERVAL_MS) continue; + if (bestIdx == outCount || p->pingMs < bestPing) { + bestIdx = i; + bestPing = p->pingMs; + } + } + if (bestIdx == outCount) break; // nothing eligible + + // Mark queried so it isn't picked again this tick, and queue the send. + struct sockaddr_storage ep; + if (Node_ConnListenEndpoint(outConns[bestIdx], &ep)) { + discovered_peer_t* p = Discovery_FindPtr(disc, &ep); + if (p) p->lastQueryMs = now; + } + toQuery[toQueryCount++] = outConns[bestIdx]; + queries++; + } + } + + // 5. Connect: pick REACHABLE, not-currently-connected, cooled-down peers with the lowest ping + // until we reach the target connection count. + if (outCount < (size_t)DISCOVERY_TARGET_CONNECTIONS) { + size_t slots = (size_t)DISCOVERY_TARGET_CONNECTIONS - outCount; + for (size_t s = 0; s < slots && toConnectCount < MAX_CONS; ++s) { + discovered_peer_t* best = NULL; + 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 (p->state != DISCOVERY_STATE_REACHABLE) continue; + if (p->lastConnectMs != 0 && (now - p->lastConnectMs) < DISCOVERY_CONNECT_RETRY_MS) continue; + int already = 0; + for (size_t j = 0; j < outEpCount; ++j) { + if (Discovery_AddrEqual(&p->addr, &outEndpoints[j])) { already = 1; break; } + } + if (already) continue; + if (!best || p->pingMs < best->pingMs) best = p; + } + if (!best) break; + + best->lastConnectMs = now; // reserve so it isn't picked again this tick + char ip[INET6_ADDRSTRLEN]; + unsigned short port = 0; + if (Discovery_AddrToIpPort(&best->addr, ip, sizeof(ip), &port) && port != 0) { + strncpy(toConnect[toConnectCount].ip, ip, INET6_ADDRSTRLEN - 1); + toConnect[toConnectCount].ip[INET6_ADDRSTRLEN - 1] = '\0'; + toConnect[toConnectCount].port = port; + toConnectCount++; + } + } + } + + pthread_mutex_unlock(&disc->lock); + + // Execute the deferred network actions outside the lock. + for (size_t i = 0; i < toQueryCount; ++i) { + Node_SendPacket(disc->node, toQuery[i], PACKET_TYPE_GET_PEERS, NULL, 0); + } + for (size_t i = 0; i < toConnectCount; ++i) { + printf("NodeDiscovery: connecting to discovered peer %s:%u\n", toConnect[i].ip, toConnect[i].port); + (void)Node_ConnectPeer(disc->node, toConnect[i].ip, toConnect[i].port); + } +} + +// ---- diagnostics ------------------------------------------------------------------------- + +void NodeDiscovery_PrintPeers(node_discovery_t* disc) { + if (!disc) { + printf("NodeDiscovery: not active\n"); + return; + } + + static const char* stateNames[] = { "NEW", "PINGED", "REACHABLE", "CONNECTED", "UNREACHABLE" }; + + pthread_mutex_lock(&disc->lock); + size_t n = DynArr_size(disc->peers); + printf("Known peers (%zu):\n", n); + for (size_t i = 0; i < n; ++i) { + discovered_peer_t* p = (discovered_peer_t*)DynArr_at(disc->peers, i); + char ip[INET6_ADDRSTRLEN] = {0}; + unsigned short port = 0; + Discovery_AddrToIpPort(&p->addr, ip, sizeof(ip), &port); + const char* stateStr = (p->state <= DISCOVERY_STATE_UNREACHABLE) ? stateNames[p->state] : "?"; + if (p->pingMs == UINT64_MAX) { + printf(" %-46s hop=%u state=%-11s ping=--\n", ip, p->hop, stateStr); + } else { + printf(" %-46s hop=%u state=%-11s ping=%" PRIu64 "ms\n", ip, p->hop, stateStr, p->pingMs); + } + (void)port; // port is part of ip endpoint identity; shown via connect logs + } + pthread_mutex_unlock(&disc->lock); +} diff --git a/src/udpd/udpnode.c b/src/udpd/udpnode.c new file mode 100644 index 0000000..37fd8ed --- /dev/null +++ b/src/udpd/udpnode.c @@ -0,0 +1,379 @@ +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include + +typedef struct { + udp_node_t* node; + int sockFd; +} udprecv_thread_args_t; + +// Send a raw PING packet (nonce already chosen) to dest. +static void UdpNode_SendRawPing(udp_node_t* node, uint64_t nonce, const struct sockaddr_storage* dest) { + unsigned char buf[UDP_PING_WIRE_SIZE]; + buf[0] = (unsigned char)UDP_PACKET_TYPE_PING; + memcpy(buf + 1, &nonce, sizeof(nonce)); + + int sock = -1; + socklen_t addrLen = 0; + + if (dest->ss_family == AF_INET6 && node->sockFd >= 0) { + sock = node->sockFd; + addrLen = sizeof(struct sockaddr_in6); + } else if (dest->ss_family == AF_INET && node->sockFdV4 >= 0) { + sock = node->sockFdV4; + addrLen = sizeof(struct sockaddr_in); + } + + if (sock < 0) { + return; + } + + sendto(sock, buf, sizeof(buf), 0, (const struct sockaddr*)dest, addrLen); +} + +static void UdpNode_HandlePacket(udp_node_t* node, int fromSock, + const unsigned char* buf, ssize_t n, + const struct sockaddr_storage* from) { + if (n < 1) { + return; + } + + udp_packet_type_t type = (udp_packet_type_t)buf[0]; + + switch (type) { + case UDP_PACKET_TYPE_PING: { + if (n < UDP_PING_WIRE_SIZE) { + return; + } + uint64_t nonce; + memcpy(&nonce, buf + 1, sizeof(nonce)); + + // Build and send PONG + unsigned char reply[UDP_PONG_WIRE_SIZE]; + reply[0] = (unsigned char)UDP_PACKET_TYPE_PONG; + memcpy(reply + 1, &nonce, sizeof(nonce)); + int32_t protoVer = (int32_t)PROTO_VERSION; + memcpy(reply + 1 + sizeof(nonce), &protoVer, sizeof(protoVer)); + + socklen_t addrLen = (from->ss_family == AF_INET6) + ? sizeof(struct sockaddr_in6) + : sizeof(struct sockaddr_in); + sendto(fromSock, reply, sizeof(reply), 0, (const struct sockaddr*)from, addrLen); + break; + } + + case UDP_PACKET_TYPE_PONG: { + if (n < UDP_PONG_WIRE_SIZE) { + return; + } + uint64_t nonce; + int32_t protoVer; + memcpy(&nonce, buf + 1, sizeof(nonce)); + memcpy(&protoVer, buf + 1 + sizeof(nonce), sizeof(protoVer)); + + bool found = false; + uint64_t rttMs = 0; + pthread_mutex_lock(&node->pingsMutex); + for (int i = 0; i < UDP_MAX_PENDING_PINGS; i++) { + if (node->pendingPings[i].active && node->pendingPings[i].nonce == nonce) { + uint64_t nowMs = get_current_time_ms(); + rttMs = (nowMs >= node->pendingPings[i].lastSentMs) + ? (nowMs - node->pendingPings[i].lastSentMs) + : 0; + node->pendingPings[i].active = false; + found = true; + break; + } + } + pthread_mutex_unlock(&node->pingsMutex); + + if (found && node->on_pong) { + node->on_pong(node, from, nonce, (int)protoVer, rttMs, node->callbackUser); + } + break; + } + + default: + break; + } +} + +static void* UdpNode_RecvThreadProc(void* arg) { + udprecv_thread_args_t* args = (udprecv_thread_args_t*)arg; + udp_node_t* node = args->node; + int sock = args->sockFd; + free(args); + + unsigned char buf[1500]; + + while (node->isRunning) { + struct sockaddr_storage from; + socklen_t fromLen = sizeof(from); + ssize_t n = recvfrom(sock, buf, sizeof(buf), 0, + (struct sockaddr*)&from, &fromLen); + if (n < 1) { + if (!node->isRunning) { + break; + } + // Transient error — keep going + continue; + } + UdpNode_HandlePacket(node, sock, buf, n, &from); + } + + return NULL; +} + +static void* UdpNode_RetryThreadProc(void* arg) { + udp_node_t* node = (udp_node_t*)arg; + + struct { + uint64_t nonce; + struct sockaddr_storage dest; + } timedOut[UDP_MAX_PENDING_PINGS]; + + while (node->isRunning) { + sleep_for_milliseconds(100); + + int timedOutCount = 0; + + pthread_mutex_lock(&node->pingsMutex); + uint64_t now = get_current_time_ms(); + for (int i = 0; i < UDP_MAX_PENDING_PINGS; i++) { + pending_ping_t* p = &node->pendingPings[i]; + if (!p->active) { + continue; + } + if (now - p->lastSentMs < UDP_PING_RETRY_INTERVAL_MS) { + continue; + } + if (p->retries >= UDP_PING_MAX_RETRIES) { + timedOut[timedOutCount].nonce = p->nonce; + timedOut[timedOutCount].dest = p->dest; + timedOutCount++; + p->active = false; + } else { + UdpNode_SendRawPing(node, p->nonce, &p->dest); + p->retries++; + p->lastSentMs = now; + } + } + pthread_mutex_unlock(&node->pingsMutex); + + for (int i = 0; i < timedOutCount; i++) { + if (node->on_ping_timeout) { + node->on_ping_timeout(node, &timedOut[i].dest, + timedOut[i].nonce, node->callbackUser); + } + } + } + + return NULL; +} + +int UdpNode_Init(udp_node_t* node, uint16_t port) { + if (!node) { + return -1; + } + + memset(node, 0, sizeof(*node)); + node->sockFd = -1; + node->sockFdV4 = -1; + + int opt = 1; + + // IPv6 (pure, not dual-stack) + int fd6 = socket(AF_INET6, SOCK_DGRAM, 0); + if (fd6 >= 0) { + setsockopt(fd6, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)); + int v6only = 1; + setsockopt(fd6, IPPROTO_IPV6, IPV6_V6ONLY, &v6only, sizeof(v6only)); + + struct sockaddr_in6 a6; + memset(&a6, 0, sizeof(a6)); + a6.sin6_family = AF_INET6; + a6.sin6_port = htons(port); + a6.sin6_addr = in6addr_any; + + if (bind(fd6, (struct sockaddr*)&a6, sizeof(a6)) == 0) { + node->sockFd = fd6; + } else { + close(fd6); + } + } + + // IPv4 + int fd4 = socket(AF_INET, SOCK_DGRAM, 0); + if (fd4 >= 0) { + setsockopt(fd4, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)); + + struct sockaddr_in a4; + memset(&a4, 0, sizeof(a4)); + a4.sin_family = AF_INET; + a4.sin_port = htons(port); + a4.sin_addr.s_addr = INADDR_ANY; + + if (bind(fd4, (struct sockaddr*)&a4, sizeof(a4)) == 0) { + node->sockFdV4 = fd4; + } else { + close(fd4); + } + } + + if (node->sockFd < 0 && node->sockFdV4 < 0) { + return -1; + } + + pthread_mutex_init(&node->pingsMutex, NULL); + return 0; +} + +void UdpNode_SetCallbacks(udp_node_t* node, + void (*on_pong)(udp_node_t*, const struct sockaddr_storage*, uint64_t, int, uint64_t, void*), + void (*on_ping_timeout)(udp_node_t*, const struct sockaddr_storage*, uint64_t, void*), + void* user) { + if (!node) { + return; + } + node->on_pong = on_pong; + node->on_ping_timeout = on_ping_timeout; + node->callbackUser = user; +} + +int UdpNode_Start(udp_node_t* node) { + if (!node || node->isRunning) { + return -1; + } + if (node->sockFd < 0 && node->sockFdV4 < 0) { + return -1; + } + + node->isRunning = 1; + int anyStarted = 0; + + if (node->sockFd >= 0) { + udprecv_thread_args_t* args = (udprecv_thread_args_t*)malloc(sizeof(*args)); + if (args) { + args->node = node; + args->sockFd = node->sockFd; + if (pthread_create(&node->recvThreadV6, NULL, UdpNode_RecvThreadProc, args) == 0) { + anyStarted = 1; + } else { + free(args); + } + } + } + + if (node->sockFdV4 >= 0) { + udprecv_thread_args_t* args = (udprecv_thread_args_t*)malloc(sizeof(*args)); + if (args) { + args->node = node; + args->sockFd = node->sockFdV4; + if (pthread_create(&node->recvThreadV4, NULL, UdpNode_RecvThreadProc, args) == 0) { + anyStarted = 1; + } else { + free(args); + } + } + } + + if (pthread_create(&node->retryThread, NULL, UdpNode_RetryThreadProc, node) == 0) { + anyStarted = 1; + } + + if (!anyStarted) { + node->isRunning = 0; + return -1; + } + + return 0; +} + +void UdpNode_Stop(udp_node_t* node) { + if (!node || !node->isRunning) { + return; + } + + node->isRunning = 0; + + // Close sockets to unblock recvfrom in receive threads + if (node->sockFd >= 0) { + int fd = node->sockFd; + node->sockFd = -1; + close(fd); + } + if (node->sockFdV4 >= 0) { + int fd = node->sockFdV4; + node->sockFdV4 = -1; + close(fd); + } + + pthread_join(node->recvThreadV6, NULL); + pthread_join(node->recvThreadV4, NULL); + pthread_join(node->retryThread, NULL); +} + +void UdpNode_Destroy(udp_node_t* node) { + if (!node) { + return; + } + + if (node->sockFd >= 0) { + close(node->sockFd); + node->sockFd = -1; + } + if (node->sockFdV4 >= 0) { + close(node->sockFdV4); + node->sockFdV4 = -1; + } + + pthread_mutex_destroy(&node->pingsMutex); +} + +int UdpNode_SendPing(udp_node_t* node, const struct sockaddr_storage* dest) { + if (!node || !dest) { + return -1; + } + if (node->sockFd < 0 && node->sockFdV4 < 0) { + return -1; + } + + uint64_t nonce = random_eight_byte(); + uint64_t now = get_current_time_ms(); + + pthread_mutex_lock(&node->pingsMutex); + + int slot = -1; + for (int i = 0; i < UDP_MAX_PENDING_PINGS; i++) { + if (!node->pendingPings[i].active) { + slot = i; + break; + } + } + + if (slot < 0) { + pthread_mutex_unlock(&node->pingsMutex); + return -1; + } + + node->pendingPings[slot].nonce = nonce; + node->pendingPings[slot].dest = *dest; + node->pendingPings[slot].lastSentMs = now; + node->pendingPings[slot].retries = 0; + node->pendingPings[slot].active = true; + + pthread_mutex_unlock(&node->pingsMutex); + + UdpNode_SendRawPing(node, nonce, dest); + return 0; +}