#ifndef HYPERSWARM_DHT_CONNECTOR_H
#define HYPERSWARM_DHT_CONNECTOR_H
// DhtConnector — orchestrates the full DHT connection flow:
// find_peer → Noise IK handshake → holepunch → UDX stream.
// Wire-compatible with holepunchto/hyperdht connect.js.
#include "dht_messages.h"
#include "holepuncher.h"
#include "hyperdht.h"
#include "noise_ik.h"
#include <cstdint>
#include <cstring>
#include <functional>
#include <vector>
extern "C" {
#include <udx.h>
}
// ---- Connection state ----
enum DhtConnectionState {
DHT_CONN_IDLE = 0,
DHT_CONN_FINDING, // Looking up peer via find_peer.
DHT_CONN_HANDSHAKING, // Noise IK handshake through relay.
DHT_CONN_HOLEPUNCHING, // Exchanging holepunch payloads.
DHT_CONN_PUNCHING, // Active UDP hole-punch probing.
DHT_CONN_CONNECTED, // UDX stream established.
DHT_CONN_FAILED, // Connection failed.
};
// ---- Per-connection state ----
struct DhtConnection {
uint32_t id;
DhtConnectionState state;
bool active;
bool is_initiator;
// Remote peer identity.
uint8_t remote_public_key[32];
// Discovery result.
IPv4Address relay_address;
bool has_relay;
// Server's direct address (from handshake reply's peerAddress field).
// Matches JS: c.connect.serverAddress = hs.peerAddress || to
IPv4Address server_address;
bool has_server_address;
// Find peer query ID — only match on_peer_found from THIS query,
// not from topic lookups which return the wrong relay node.
uint32_t find_peer_qid;
// Handshake.
uint32_t handshake_id;
uint8_t holepunch_secret[32];
uint8_t tx_key[32];
uint8_t rx_key[32];
uint8_t handshake_hash[64];
// Remote's NoisePayload from handshake.
dht_msg::NoisePayload remote_payload;
bool has_remote_payload;
// UDX info from remote.
// The peer's UDX stream id, taken from the NoisePayload it sent. That payload
// carries the HANDSHAKE id (hyperdht.cpp: `payload.udx.id = hs->id`), so this
// pairs with `handshake_id` on our side — see udx_transport's _dht_on_connected,
// which uses handshake_id as OUR local stream id for exactly that reason.
//
// There is deliberately no `local_stream_id` here. There used to be, assigned
// from a counter based at 100 and advertised by a payload encoder that was never
// called — so it named a stream nothing ever used, while the log printed it as
// though it were the real one. That decoy is what made a stale stream id hard to
// pin in hyper-games #15; the wire has exactly one source of truth and it is the
// handshake id.
uint32_t remote_stream_id;
uint32_t remote_seq;
// Holepunch state.
Holepuncher puncher;
uint32_t holepunch_round;
IPv4Address punched_address;
bool punched;
bool puncher_started;
// Firewall info.
uint32_t local_firewall;
uint32_t remote_firewall;
// Timestamps.
uint64_t started_at;
uint64_t last_activity;
// Winning birthday socket (transferred from Holepuncher).
// Non-null if connection was established via birthday attack.
udx_socket_t *punch_socket;
// When the RELAY stage began — handshake or holepunch coordination through a
// DHT node. 0 until it does. See DHT_RELAY_TIMEOUT_MS.
uint64_t relay_since;
// Error info.
const char *error;
};
static constexpr int DHT_MAX_CONNECTIONS = 16;
static constexpr uint64_t DHT_CONNECT_TIMEOUT_MS = 30000; // 30 seconds.
// ⛔ THE RELAY STAGE GETS JS'S DEADLINE, NOT THE BACKSTOP.
//
// hyperdht arms `c.relayTimeout = setTimeout(onabort, 15000)` (connect.js:812)
// when it starts talking to a relay, and abandons the attempt there. Ours held
// every stage to the same 30s wall clock, so a peer that answered the DHT and
// then went quiet occupied one of SWARM_MAX_PARALLEL = 3 outbound slots for
// twice as long as the reference — and a topic carrying stale announce records
// is exactly a supply of those.
//
// Measured from when the RELAY stage began, not from when the attempt started:
// JS's timer starts with the relay, and a connection that spent twelve seconds
// in find_peer has not used any of its relay budget.
//
// ⚠ The OTHER row of PARITY §11.1's table — JS's 10s passiveConnectTimeout for
// a peer whose firewall is OPEN — is NOT implemented here. It needs a firewall
// state this connector does not track at that point, and inventing a mapping
// for it would be guessing at parity rather than achieving it.
static constexpr uint64_t DHT_RELAY_TIMEOUT_MS = 15000; // JS connect.js:812.
static constexpr uint64_t DHT_KEEPALIVE_INTERVAL_MS = 10000; // 10 seconds.
static constexpr uint64_t DHT_KEEPALIVE_TIMEOUT_MS = 30000; // 30 seconds.
static constexpr uint64_t RANDOM_PUNCH_COOLDOWN_MS = 20000; // 20 seconds between random punches.
// How long an initiator conn must sit in FINDING/HANDSHAKING with no server
// address before the dedup will treat it as STUCK and prefer an incoming
// responder instead. The stall this recovers from (hyper-fleeky-dev #93) is
// PERMANENT — a find_peer reply that never carries the peer's server address —
// so any dwell well under DHT_CONNECT_TIMEOUT_MS preserves it. Without a dwell
// the predicate also matches a healthy conn milliseconds old, which is how it
// deadlocked two peers that were connecting perfectly well (hyper-games #15).
static constexpr uint64_t STUCK_INITIATOR_MIN_MS = 3000; // 30x the observed healthy path.
static constexpr int RANDOM_PUNCH_LIMIT = 1; // Max concurrent random punches.
// Punch statistics — matches JS dht.stats.punches.
struct PunchStats {
uint32_t total;
uint32_t consistent_ok;
uint32_t random_ok;
uint32_t open_ok;
uint32_t timeouts;
};
// Snapshot of a single connection for diagnostics (thread-safe copy).
struct DhtConnectionInfo {
uint32_t id;
DhtConnectionState state;
bool is_initiator;
uint8_t remote_public_key[32];
IPv4Address relay_address;
bool has_relay;
IPv4Address server_address;
bool has_server_address;
IPv4Address punched_address;
bool punched;
uint32_t holepunch_round;
uint32_t local_firewall;
uint32_t remote_firewall;
uint64_t started_at;
uint64_t elapsed_ms;
const char *error;
// Remote's advertised addresses (from NoisePayload).
IPv4Address remote_addresses[8];
int remote_address_count;
};
// ---- DhtConnector ----
class DhtConnector {
public:
DhtConnector();
~DhtConnector();
// Initialize with the DHT stack and UDP socket.
void init(HyperDht *dht, udx_socket_t *socket, uv_loop_t *loop);
// Initiate a connection to a remote peer by public key.
// Returns connection ID, or 0 on failure.
uint32_t connect(const uint8_t *remote_public_key);
// Start listening for incoming DHT connections.
void listen();
// Stop listening.
void unlisten();
// Tick — advance all connection state machines.
void tick(uint64_t now_ms);
// Handle an incoming holepunch request from the DHT (PEER_HOLEPUNCH command).
void on_holepunch_message(const DhtRequest &req);
// Handle a holepunch RPC response (reply to our outgoing PEER_HOLEPUNCH).
// tag is 0x10000 | conn_id. Returns true if handled.
bool on_holepunch_response(uint32_t tag, const uint8_t *value, size_t value_len,
const IPv4Address &from);
// Handle a raw UDP probe received on the socket.
void on_probe_received(const IPv4Address &from);
// ⛔ THE FIND_PEER QUERY EXHAUSTED. Fail every connection still waiting on
// it, instead of holding the slot until DHT_CONNECT_TIMEOUT_MS.
//
// JS has NO overall connect timer: its find_peer stage ends when the
// iterative query itself finishes (connect.js:381 `for await (const data of
// c.query)`), so a peer that is simply not there fails as fast as the DHT
// can say so. Ours waited the full 30s wall clock for the same answer,
// which with SWARM_MAX_PARALLEL = 3 parks every outbound slot on peers the
// DHT has already reported nothing for. Stale announce records — which we
// produce, see defects A and B — are exactly that case.
void on_find_peer_complete(uint32_t query_id);
// Read-only lookup, for diagnostics and gates. Does not extend the
// connection's life: the pointer is into the fixed array and is valid only
// until the next connect/free.
const DhtConnection *find_connection(uint32_t id) const;
// A DOOR FOR GATES, and named so it cannot be mistaken for anything else.
//
// Reaching HOLEPUNCHING legitimately needs a relay, a live peer and a
// completed Noise handshake — a whole swarm, to test one branch that runs
// when a relay answers with something unreadable. The alternative to this
// door is an ungated abort path, and this repo has already paid for guards
// that had no test: deleting one then looks like a fix (§8.3).
//
// It moves ONE connection's state and nothing else. It is not used outside
// test/.
bool set_state_for_gate(uint32_t conn_id, DhtConnectionState state);
// Get the local IPv4 addresses we advertise in handshakes.
void set_local_addresses(const IPv4Address *addrs, int count);
// ---- Callbacks ----
// Called when a connection is fully established (UDX-ready).
// Provides: connection ID, remote public key, punched address,
// remote UDX stream ID, tx/rx keys for NoiseStream.
std::function<void(uint32_t conn_id, const DhtConnection &conn)> on_connected;
// Called when a connection fails.
std::function<void(uint32_t conn_id, const uint8_t *remote_pk, const char *error)> on_failed;
// Called when an incoming connection request is received (if listening).
std::function<void(uint32_t conn_id, const uint8_t *remote_pk)> on_incoming;
bool is_listening() const { return _listening; }
// Get punch statistics.
const PunchStats &punch_stats() const { return _stats; }
// Get a snapshot of all active connections for diagnostics.
// Returns the number of entries written to `out` (max DHT_MAX_CONNECTIONS).
int get_connections_info(DhtConnectionInfo *out, int max_count, uint64_t now_ms) const;
private:
// Advance a single connection.
void _tick_connection(DhtConnection *conn, uint64_t now_ms);
// Move connection to next state after handshake completes.
void _on_handshake_complete(DhtConnection *conn, const DhtHandshake &hs);
// Start the holepunch exchange phase.
void _start_holepunch(DhtConnection *conn);
// Send a holepunch payload through the DHT relay.
void _send_holepunch(DhtConnection *conn);
// Process a received holepunch payload.
void _process_holepunch(DhtConnection *conn, const dht_msg::HolepunchPayload &payload);
// Finalize connection — create UDX stream.
void _finalize_connection(DhtConnection *conn, const IPv4Address &addr);
// Connection management.
DhtConnection *_alloc_connection(bool initiator, const uint8_t *remote_pk);
// Fail one attempt immediately rather than at the wall clock.
void _fail_connection(DhtConnection *conn, const char *reason);
DhtConnection *_find_connection(uint32_t id);
DhtConnection *_find_connection_by_pk(const uint8_t *pk);
void _free_connection(DhtConnection *conn);
// State.
HyperDht *_dht;
udx_socket_t *_socket;
uv_loop_t *_loop;
bool _listening;
DhtConnection _connections[DHT_MAX_CONNECTIONS];
uint32_t _conn_id_counter;
IPv4Address _local_addresses[HP_MAX_ADDRESSES];
int _local_address_count;
uint32_t _local_firewall;
uint64_t _now_ms;
// External IP tracking for LAN detection.
void _update_external_ip();
bool _is_same_lan(const DhtConnection *conn) const;
bool _find_lan_address(const DhtConnection *conn, IPv4Address &out) const;
static bool _is_private_ip(const uint8_t *ip);
IPv4Address _our_external_ip;
bool _has_external_ip;
// Random punch throttling — matches JS _randomPunches / _lastRandomPunch.
PunchStats _stats;
uint64_t _last_random_punch;
int _random_punch_count;
};
#endif // HYPERSWARM_DHT_CONNECTOR_H