#include "dht_connector.h"
#include "debug_log.h"
#include <sodium.h>
// ---- Construction ----
DhtConnector::DhtConnector()
: _dht(nullptr), _socket(nullptr), _loop(nullptr),
_listening(false), _conn_id_counter(0),
_local_address_count(0), _local_firewall(FIREWALL_UNKNOWN), _now_ms(0),
_has_external_ip(false),
_last_random_punch(0), _random_punch_count(0) {
memset(&_stats, 0, sizeof(_stats));
for (int i = 0; i < DHT_MAX_CONNECTIONS; i++) {
_connections[i].active = false;
}
}
DhtConnector::~DhtConnector() {
for (int i = 0; i < DHT_MAX_CONNECTIONS; i++) {
if (_connections[i].active) {
_connections[i].puncher.stop();
_connections[i].active = false;
}
}
}
void DhtConnector::init(HyperDht *dht, udx_socket_t *socket, uv_loop_t *loop) {
_dht = dht;
_socket = socket;
_loop = loop;
// Register HyperDht callbacks.
_dht->on_peer_found = [this](uint32_t query_id, const FoundPeer &peer) {
DHT_LOG("[DhtConnector] on_peer_found: qid=%u pk=%02x%02x%02x%02x relay=%d.%d.%d.%d:%d\n",
query_id,
peer.public_key[0], peer.public_key[1], peer.public_key[2], peer.public_key[3],
peer.relay_node.ip[0], peer.relay_node.ip[1], peer.relay_node.ip[2], peer.relay_node.ip[3],
peer.relay_node.port);
// Match query to connection.
// CRITICAL: Only match results from the find_peer query we started,
// NOT from topic lookups. Topic lookup relay nodes don't have
// Router forwarding entries for hash(pk) — only find_peer results
// come from nodes that stored the self-announce.
for (int i = 0; i < DHT_MAX_CONNECTIONS; i++) {
DhtConnection *conn = &_connections[i];
if (!conn->active) continue;
if (conn->state != DHT_CONN_FINDING) continue;
// Check if this peer matches our target.
if (memcmp(conn->remote_public_key, peer.public_key, 32) != 0) continue;
// Only accept results from our find_peer query.
if (conn->find_peer_qid != query_id) {
DHT_LOG("[DhtConnector] SKIP: qid mismatch (conn find_peer_qid=%u, got qid=%u)\n",
conn->find_peer_qid, query_id);
continue;
}
DHT_LOG("[DhtConnector] MATCHED conn %u → handshake via relay %d.%d.%d.%d:%d\n",
conn->id,
peer.relay_node.ip[0], peer.relay_node.ip[1], peer.relay_node.ip[2], peer.relay_node.ip[3],
peer.relay_node.port);
// Found our peer — use relay_node (the DHT node that has the
// forwarding entry). This is the `data.from` in JS connect flow.
// We send PEER_HANDSHAKE to this node and it routes to the server.
conn->relay_address = peer.relay_node;
conn->has_relay = true;
// Initiate handshake through the relay node.
conn->handshake_id = _dht->connect(conn->remote_public_key, conn->relay_address);
conn->state = DHT_CONN_HANDSHAKING;
if (conn->relay_since == 0) conn->relay_since = _now_ms;
conn->last_activity = 0; // Reset timeout.
break;
}
};
_dht->on_handshake_complete = [this](uint32_t handshake_id, const DhtHandshake &hs) {
DHT_LOG("[DhtConnector] on_handshake_complete: hs_id=%u has_server=%d server=%d.%d.%d.%d:%d\n",
handshake_id, hs.has_server_address,
hs.server_address.ip[0], hs.server_address.ip[1],
hs.server_address.ip[2], hs.server_address.ip[3], hs.server_address.port);
for (int i = 0; i < DHT_MAX_CONNECTIONS; i++) {
DhtConnection *conn = &_connections[i];
if (!conn->active) continue;
if (conn->state != DHT_CONN_HANDSHAKING) continue;
if (conn->handshake_id != handshake_id) continue;
_on_handshake_complete(conn, hs);
break;
}
};
_dht->on_handshake_request = [this](uint32_t handshake_id, const uint8_t *remote_pk,
const dht_msg::NoisePayload &payload) {
DHT_LOG("[DhtConnector] on_handshake_request: hs_id=%u listening=%d pk=%02x%02x%02x%02x\n",
handshake_id, _listening,
remote_pk ? remote_pk[0] : 0, remote_pk ? remote_pk[1] : 0,
remote_pk ? remote_pk[2] : 0, remote_pk ? remote_pk[3] : 0);
if (!_listening) return;
// ---- Connection dedup (matches JS _handleServerConnection) ----
// JS: const existing = this._allConnections.get(conn.remotePublicKey)
DhtConnection *existing = _find_connection_by_pk(remote_pk);
if (existing) {
// JS: expectedInitiator = b4a.compare(conn.publicKey, conn.remotePublicKey) > 0
int cmp = memcmp(_dht->public_key(), remote_pk, 32);
bool we_should_initiate = (cmp > 0);
// JS: existingIsOutdated = existing.rawBytesRead > 0 && existing.rawBytesWritten > 0
bool existing_is_outdated = (existing->state == DHT_CONN_CONNECTED);
// Same-NAT / simultaneous-connect recovery (hyper-fleeky-dev #93):
// when both peers are behind one NAT, our elected-initiator conn can
// STALL — its find_peer reply never carries the peer's server address
// (has_server_address stays false), so it never holepunches. Meanwhile
// THIS incoming responder handshake is fully formed (completed Noise +
// client_address from the relay's peerAddress + the peer's LAN payload)
// and CAN connect. So if our existing conn is a stuck initiator, prefer
// the incoming instead of clinging to a conn that can never punch.
//
// THE DWELL IS LOAD-BEARING, and its absence cost ~15s on every
// connection (hyper-games #15). This is the ONLY term that lets the
// pk-elected initiator drop its own outgoing conn, and the rule above it
// is asymmetric ON PURPOSE: exactly one peer keeps its outgoing, the
// other adopts the incoming, and that is what makes the two sides agree
// on one connection. Firing this on a conn milliseconds old made BOTH
// peers adopt simultaneously — two responders, zero initiators — so each
// side addressed a UDX stream the other had just freed. Both sent a Noise
// header, neither received one, both closed -110 after ~15s, and only the
// retry connected. Layer 12 measured 19.4s one run and never the next.
//
// A conn that is merely YOUNG is not stuck; it is still working. #93's
// stall never resolves, so waiting costs it nothing and costs the healthy
// path everything.
const uint64_t existing_age_ms =
(existing->started_at == 0 || _now_ms < existing->started_at)
? 0 : (_now_ms - existing->started_at);
bool existing_stuck_initiator = existing->is_initiator &&
!existing->has_server_address &&
(existing->state == DHT_CONN_FINDING ||
existing->state == DHT_CONN_HANDSHAKING) &&
existing_age_ms >= STUCK_INITIATOR_MIN_MS;
// New incoming: isInitiator = false (we're responding).
// JS: keepNew = existingIsOutdated || (expectedInitiator === conn.isInitiator)
// For incoming: keepNew = existingIsOutdated || !we_should_initiate
bool keep_new = existing_is_outdated || !we_should_initiate ||
existing_stuck_initiator;
if (!keep_new) {
DHT_LOG("[DhtConnector] on_handshake_request: DEDUP — rejecting incoming (keep existing)\n");
return;
}
if (existing_stuck_initiator) {
DHT_LOG("[DhtConnector] on_handshake_request: DEDUP — existing initiator conn %u stuck (no server addr); taking incoming responder instead\n", existing->id);
}
// Keep incoming, destroy existing.
DHT_LOG("[DhtConnector] on_handshake_request: DEDUP — replacing existing conn %u\n", existing->id);
if (on_failed) on_failed(existing->id, existing->remote_public_key, "duplicate connection (replaced by incoming)");
_free_connection(existing);
}
// Incoming connection — allocate state.
DhtConnection *conn = _alloc_connection(false, remote_pk);
if (!conn) return;
conn->handshake_id = handshake_id;
conn->remote_payload = payload;
conn->has_remote_payload = true;
// Extract UDX info from remote payload.
if (payload.has_udx) {
conn->remote_stream_id = payload.udx.id;
conn->remote_seq = payload.udx.seq;
}
// Extract firewall info.
conn->remote_firewall = payload.firewall;
if (on_incoming) {
on_incoming(conn->id, remote_pk);
}
// For responder: stay in HANDSHAKING state and wait for
// on_handshake_complete which provides the client_address
// (from relay's peerAddress). This avoids premature holepunch failure.
conn->state = DHT_CONN_HANDSHAKING;
if (conn->relay_since == 0) conn->relay_since = _now_ms;
};
}
// ---- Public API ----
uint32_t DhtConnector::connect(const uint8_t *remote_public_key) {
// Self-connection prevention: don't connect to our own public key.
// JS hyperswarm checks this in PeerDiscovery before initiating.
if (memcmp(remote_public_key, _dht->public_key(), 32) == 0) {
DHT_LOG("[DhtConnector] connect: SKIP self-connection (our own pk)\n");
return 0;
}
DhtConnection *conn = _alloc_connection(true, remote_public_key);
if (!conn) return 0;
// Start peer discovery.
conn->state = DHT_CONN_FINDING;
conn->find_peer_qid = _dht->find_peer(remote_public_key);
return conn->id;
}
void DhtConnector::listen() {
_listening = true;
_dht->listen();
}
void DhtConnector::unlisten() {
_listening = false;
}
void DhtConnector::set_local_addresses(const IPv4Address *addrs, int count) {
_local_address_count = (count > HP_MAX_ADDRESSES) ? HP_MAX_ADDRESSES : count;
for (int i = 0; i < _local_address_count; i++) {
_local_addresses[i] = addrs[i];
}
}
// ---- Tick ----
void DhtConnector::tick(uint64_t now_ms) {
_now_ms = now_ms;
// Update local firewall from DhtRpc's NatSampler.
uint32_t detected = _dht->rpc()->local_firewall();
if (detected != FIREWALL_UNKNOWN) {
_local_firewall = detected;
}
for (int i = 0; i < DHT_MAX_CONNECTIONS; i++) {
if (_connections[i].active) {
_tick_connection(&_connections[i], now_ms);
}
}
}
void DhtConnector::_tick_connection(DhtConnection *conn, uint64_t now_ms) {
if (conn->started_at == 0) conn->started_at = now_ms;
if (conn->last_activity == 0) conn->last_activity = now_ms;
// Timeout check. The relay stage has JS's shorter deadline; everything else
// keeps the 30s backstop, which is all the find_peer stage needs now that it
// is released when the query exhausts (§11.2).
const bool in_relay = (conn->state == DHT_CONN_HANDSHAKING ||
conn->state == DHT_CONN_HOLEPUNCHING);
const bool relay_expired = in_relay && conn->relay_since != 0 &&
now_ms - conn->relay_since > DHT_RELAY_TIMEOUT_MS;
if (conn->state != DHT_CONN_CONNECTED &&
(relay_expired || now_ms - conn->started_at > DHT_CONNECT_TIMEOUT_MS)) {
conn->state = DHT_CONN_FAILED;
conn->error = relay_expired ? "relay timeout" : "connection timeout";
_stats.timeouts++;
// Release random punch slot if applicable.
HolepunchStage pstage = conn->puncher.stage();
if ((pstage == HP_PUNCH_RANDOM || pstage == HP_PUNCH_BIRTHDAY) &&
_random_punch_count > 0) {
_random_punch_count--;
_last_random_punch = now_ms;
}
if (on_failed) on_failed(conn->id, conn->remote_public_key, conn->error);
_free_connection(conn);
return;
}
// Keep-alive for connected peers.
if (conn->state == DHT_CONN_CONNECTED) {
if (now_ms - conn->last_activity > DHT_KEEPALIVE_TIMEOUT_MS) {
conn->state = DHT_CONN_FAILED;
conn->error = "keepalive timeout";
if (on_failed) on_failed(conn->id, conn->remote_public_key, conn->error);
_free_connection(conn);
return;
}
// Periodic keepalive probes (1-byte) every 10s.
if (conn->punched && now_ms - conn->last_activity > DHT_KEEPALIVE_INTERVAL_MS) {
conn->puncher.send_probe(conn->punched_address, HP_TTL_NORMAL);
}
return;
}
// Periodic resend while waiting for coordination response.
if (conn->state == DHT_CONN_HOLEPUNCHING && !conn->puncher_started) {
if (now_ms - conn->last_activity > 2000) {
_send_holepunch(conn);
conn->last_activity = now_ms;
}
}
// Advance punching state.
if (conn->state == DHT_CONN_PUNCHING) {
if (!conn->puncher.tick(now_ms)) {
if (conn->puncher.is_connected()) {
// Punch succeeded!
DHT_LOG("[DhtConnector] tick: conn=%u punch complete via tick, addr=%d.%d.%d.%d:%d\n",
conn->id, conn->punched_address.ip[0], conn->punched_address.ip[1],
conn->punched_address.ip[2], conn->punched_address.ip[3], conn->punched_address.port);
_finalize_connection(conn, conn->punched_address);
}
// If not connected and not punching, the abort callback handles it.
}
}
}
// ---- Handshake completion ----
void DhtConnector::_on_handshake_complete(DhtConnection *conn, const DhtHandshake &hs) {
memcpy(conn->holepunch_secret, hs.holepunch_secret, 32);
memcpy(conn->tx_key, hs.tx_key, 32);
memcpy(conn->rx_key, hs.rx_key, 32);
memcpy(conn->handshake_hash, hs.hash, 64);
// Extract server address from handshake (relay's peerAddress field).
if (hs.has_server_address) {
conn->server_address = hs.server_address;
conn->has_server_address = true;
}
// For responder: extract client address from relay's peerAddress.
// This is the initiator's address as seen by the relay.
if (hs.has_client_address) {
conn->server_address = hs.client_address; // Reuse server_address for the target.
conn->has_server_address = true;
DHT_LOG("[DhtConnector] _on_handshake_complete: responder got client_address=%d.%d.%d.%d:%d\n",
hs.client_address.ip[0], hs.client_address.ip[1],
hs.client_address.ip[2], hs.client_address.ip[3], hs.client_address.port);
}
// Set relay address from handshake (needed for holepunch RPC coordination).
// Initiator already has this from on_peer_found, but responder needs it
// from the handshake — without it, _send_holepunch silently does nothing.
if (!conn->has_relay && hs.relay_address.port != 0) {
conn->relay_address = hs.relay_address;
conn->has_relay = true;
DHT_LOG("[DhtConnector] _on_handshake_complete: set relay=%d.%d.%d.%d:%d from handshake\n",
hs.relay_address.ip[0], hs.relay_address.ip[1],
hs.relay_address.ip[2], hs.relay_address.ip[3], hs.relay_address.port);
}
// Extract remote NoisePayload (has UDX stream ID, firewall, addresses).
if (hs.has_remote_payload) {
conn->remote_payload = hs.remote_payload;
conn->has_remote_payload = true;
if (hs.remote_payload.has_udx) {
conn->remote_stream_id = hs.remote_payload.udx.id;
conn->remote_seq = hs.remote_payload.udx.seq;
}
conn->remote_firewall = hs.remote_payload.firewall;
}
_start_holepunch(conn);
}
// ---- Holepunch ----
void DhtConnector::_start_holepunch(DhtConnection *conn) {
conn->state = DHT_CONN_HOLEPUNCHING;
if (conn->relay_since == 0) conn->relay_since = _now_ms;
conn->holepunch_round = 0;
conn->punch_socket = nullptr;
DHT_LOG("[DhtConnector] _start_holepunch: conn=%u initiator=%d has_payload=%d fw=%d has_server=%d server=%d.%d.%d.%d:%d\n",
conn->id, conn->is_initiator, conn->has_remote_payload,
conn->remote_firewall, conn->has_server_address,
conn->server_address.ip[0], conn->server_address.ip[1],
conn->server_address.ip[2], conn->server_address.ip[3], conn->server_address.port);
DHT_LOG("[DhtConnector] has_relay=%d relay=%d.%d.%d.%d:%d local_addrs=%d\n",
conn->has_relay,
conn->relay_address.ip[0], conn->relay_address.ip[1],
conn->relay_address.ip[2], conn->relay_address.ip[3], conn->relay_address.port,
_local_address_count);
for (int la = 0; la < _local_address_count; la++) {
DHT_LOG("[DhtConnector] local[%d]=%d.%d.%d.%d:%d\n", la,
_local_addresses[la].ip[0], _local_addresses[la].ip[1],
_local_addresses[la].ip[2], _local_addresses[la].ip[3], _local_addresses[la].port);
}
// Check if we can skip holepunch (remote is OPEN/port-forwarded).
// Matches JS: if (payload.firewall === FIREWALL.OPEN || (relayed && !remoteHolepunchable))
if (conn->has_remote_payload && conn->remote_firewall == FIREWALL_OPEN) {
// Direct connection — use first address from remote.
if (conn->remote_payload.addresses4_count > 0) {
_finalize_connection(conn, conn->remote_payload.addresses4[0]);
return;
}
// Fallback: use server address from handshake reply.
if (conn->has_server_address) {
_finalize_connection(conn, conn->server_address);
return;
}
}
// Check if we can reach the server directly (same host / not relayed).
// Matches JS: if (!relayed) { c.onsocket(socket, address.port, address.host) }
// If the handshake reply did NOT include peerAddress, the relay IS the server.
if (!conn->has_server_address && conn->has_relay) {
// Not relayed — the relay node handled the handshake directly.
// The relay address IS the server. Connect directly.
_finalize_connection(conn, conn->relay_address);
return;
}
// Check if client & server are behind the same NAT (LAN optimization).
// Matches JS: if (c.lan && relayed && clientAddress.host === serverAddress.host)
// Uses external IP comparison (NatSampler) + subnet matching on advertised addresses.
_update_external_ip();
if (conn->has_server_address && _is_same_lan(conn)) {
// Both peers share the same external IP → same NAT → same LAN.
// Find a reachable LAN address from remote's advertised addresses.
IPv4Address lan_target;
if (_find_lan_address(conn, lan_target)) {
DHT_LOG("[DhtConnector] _start_holepunch: LAN shortcut → %d.%d.%d.%d:%d\n",
lan_target.ip[0], lan_target.ip[1],
lan_target.ip[2], lan_target.ip[3], lan_target.port);
_finalize_connection(conn, lan_target);
return;
}
// Fallback: try server_address directly (might be LAN IP).
DHT_LOG("[DhtConnector] _start_holepunch: LAN fallback → server_address %d.%d.%d.%d:%d\n",
conn->server_address.ip[0], conn->server_address.ip[1],
conn->server_address.ip[2], conn->server_address.ip[3], conn->server_address.port);
_finalize_connection(conn, conn->server_address);
return;
}
// Check random punch cooldown (matching JS _randomPunches / _randomPunchInterval).
bool is_random_punch = (_local_firewall >= FIREWALL_RANDOM ||
conn->remote_firewall >= FIREWALL_RANDOM);
if (is_random_punch) {
if (_random_punch_count >= RANDOM_PUNCH_LIMIT ||
(_last_random_punch > 0 &&
_now_ms - _last_random_punch < RANDOM_PUNCH_COOLDOWN_MS)) {
// Defer — will retry on next tick when cooldown expires.
return;
}
_random_punch_count++;
}
// Initialize the holepuncher.
conn->puncher.init(_socket, _local_firewall);
conn->puncher.on_connect = [this, conn](const IPv4Address &addr) {
conn->punched_address = addr;
conn->punched = true;
_finalize_connection(conn, addr);
};
conn->puncher.on_abort = [this, conn]() {
conn->state = DHT_CONN_FAILED;
conn->error = "holepunch failed";
if (on_failed) on_failed(conn->id, conn->remote_public_key, conn->error);
_free_connection(conn);
};
// Seed puncher with all known remote addresses.
{
RemotePeerInfo rinfo;
memset(&rinfo, 0, sizeof(rinfo));
rinfo.firewall = conn->has_remote_payload ? conn->remote_firewall : FIREWALL_UNKNOWN;
rinfo.address_count = 0;
// Add server_address (relay-confirmed external address) as primary target.
if (conn->has_server_address && conn->server_address.port != 0) {
rinfo.addresses[rinfo.address_count++] = conn->server_address;
}
// Add addresses from handshake payload, deduped.
if (conn->has_remote_payload) {
for (int i = 0; i < conn->remote_payload.addresses4_count &&
rinfo.address_count < HP_MAX_ADDRESSES; i++) {
bool dup = false;
for (int j = 0; j < rinfo.address_count; j++) {
if (memcmp(rinfo.addresses[j].ip, conn->remote_payload.addresses4[i].ip, 4) == 0 &&
rinfo.addresses[j].port == conn->remote_payload.addresses4[i].port) {
dup = true; break;
}
}
if (!dup) rinfo.addresses[rinfo.address_count++] = conn->remote_payload.addresses4[i];
}
}
DHT_LOG("[DhtConnector] _start_holepunch: seeded puncher with %d remote addrs\n",
rinfo.address_count);
conn->puncher.update_remote(rinfo);
}
// Send the first holepunch message.
_send_holepunch(conn);
// Defer puncher.start() — wait for coordination response with remote addresses.
conn->puncher_started = false;
// Stay in DHT_CONN_HOLEPUNCHING until coordination response arrives.
}
void DhtConnector::_send_holepunch(DhtConnection *conn) {
// Build holepunch payload.
dht_msg::HolepunchPayload payload;
memset(&payload, 0, sizeof(payload));
payload.firewall = _local_firewall;
payload.round = conn->holepunch_round;
payload.punching = conn->puncher.is_punching();
// Include our addresses (local + NAT-detected external).
payload.addresses_count = _dht->rpc()->get_all_addresses(payload.addresses, 8);
// Encode the payload.
EncodeState ps = { 0, 0, nullptr };
dht_msg::holepunch_payload_preencode(ps, payload);
std::vector<uint8_t> pt(ps.end);
ps.buffer = pt.data();
ps.start = 0;
dht_msg::holepunch_payload_encode(ps, payload);
// Encrypt with holepunch secret.
auto encrypted = dht_msg::secure_payload_encrypt(pt.data(), pt.size(),
conn->holepunch_secret);
// Wrap in HolepunchMessage.
dht_msg::HolepunchMessage hmsg;
memset(&hmsg, 0, sizeof(hmsg));
hmsg.mode = conn->is_initiator ? 1 : 2;
hmsg.id = conn->id;
hmsg.payload = encrypted.data();
hmsg.payload_len = encrypted.size();
EncodeState hs = { 0, 0, nullptr };
dht_msg::holepunch_msg_preencode(hs, hmsg);
std::vector<uint8_t> msg_buf(hs.end);
hs.buffer = msg_buf.data();
hs.start = 0;
dht_msg::holepunch_msg_encode(hs, hmsg);
// Send via DHT-RPC as PEER_HOLEPUNCH command to the relay.
// Use tag = 0x10000 | conn->id so the response can be routed back
// to us via _on_holepunch_response() (tag=0 responses get dropped).
if (conn->has_relay) {
uint32_t hp_tag = 0x10000 | conn->id;
DHT_LOG("[DhtConnector] _send_holepunch: conn=%u round=%d tag=0x%x relay=%d.%d.%d.%d:%d\n",
conn->id, conn->holepunch_round, hp_tag,
conn->relay_address.ip[0], conn->relay_address.ip[1],
conn->relay_address.ip[2], conn->relay_address.ip[3], conn->relay_address.port);
_dht->rpc()->send_request(HYPERDHT_CMD_PEER_HOLEPUNCH, nullptr,
msg_buf.data(), msg_buf.size(),
conn->relay_address, false, hp_tag);
}
conn->holepunch_round++;
}
void DhtConnector::on_holepunch_message(const DhtRequest &req) {
// Decode the holepunch message.
EncodeState s = { 0, req.value_len, const_cast<uint8_t *>(req.value) };
dht_msg::HolepunchMessage hmsg;
if (!dht_msg::holepunch_msg_decode(s, hmsg)) return;
// Find the connection by matching the message.
// For incoming, match by source relay. For outgoing, match by our ID.
DhtConnection *conn = nullptr;
for (int i = 0; i < DHT_MAX_CONNECTIONS; i++) {
if (!_connections[i].active) continue;
if (_connections[i].state < DHT_CONN_HOLEPUNCHING) continue;
// Simple match: any active holepunching connection.
// TODO: more precise matching by relay address or connection ID.
conn = &_connections[i];
break;
}
if (!conn) return;
// Decrypt the payload.
std::vector<uint8_t> pt;
if (!dht_msg::secure_payload_decrypt(hmsg.payload, hmsg.payload_len,
conn->holepunch_secret, pt)) {
return;
}
// Decode holepunch payload.
EncodeState ps = { 0, pt.size(), pt.data() };
dht_msg::HolepunchPayload payload;
if (!dht_msg::holepunch_payload_decode(ps, payload)) return;
_process_holepunch(conn, payload);
}
// Fail one attempt NOW: report it and give the slot back. The timeout path does
// the same thing, minus the random-punch bookkeeping, which only matters for a
// connection that had reached a punching stage.
void DhtConnector::_fail_connection(DhtConnection *conn, const char *reason) {
if (!conn || !conn->active) return;
conn->state = DHT_CONN_FAILED;
conn->error = reason;
const HolepunchStage pstage = conn->puncher.stage();
if ((pstage == HP_PUNCH_RANDOM || pstage == HP_PUNCH_BIRTHDAY) &&
_random_punch_count > 0) {
_random_punch_count--;
}
DHT_LOG("[DhtConnector] conn %u failed: %s\n", conn->id, reason);
if (on_failed) on_failed(conn->id, conn->remote_public_key, conn->error);
_free_connection(conn);
}
bool DhtConnector::set_state_for_gate(uint32_t conn_id, DhtConnectionState state) {
DhtConnection *conn = _find_connection(conn_id);
if (!conn) return false;
conn->state = state;
// Entering a relay stage starts its clock, exactly as the real transitions
// do — otherwise a gate could put a connection in HOLEPUNCHING and then
// assert about a deadline that was never armed.
if ((state == DHT_CONN_HANDSHAKING || state == DHT_CONN_HOLEPUNCHING) &&
conn->relay_since == 0) {
conn->relay_since = _now_ms;
}
return true;
}
bool DhtConnector::on_holepunch_response(uint32_t tag, const uint8_t *value,
size_t value_len, const IPv4Address &from) {
// Holepunch response tags are 0x10000 | conn_id.
if ((tag & 0xFFFF0000) != 0x10000) return false;
uint32_t conn_id = tag & 0xFFFF;
DHT_LOG("[DhtConnector] on_holepunch_response: tag=0x%x conn_id=%u value_len=%zu from=%d.%d.%d.%d:%d\n",
tag, conn_id, value_len,
from.ip[0], from.ip[1], from.ip[2], from.ip[3], from.port);
if (!value || value_len == 0) {
DHT_LOG("[DhtConnector] holepunch response empty (relay couldn't reach peer)\n");
return true; // Handled, just empty.
}
// Decode the holepunch message envelope (mode=REPLY from relay).
EncodeState s = { 0, value_len, const_cast<uint8_t *>(value) };
dht_msg::HolepunchMessage hmsg;
const bool envelope_ok = dht_msg::holepunch_msg_decode(s, hmsg);
if (!envelope_ok) {
DHT_LOG("[DhtConnector] holepunch response decode failed\n");
}
DHT_LOG("[DhtConnector] holepunch response: mode=%d id=%u payload_len=%zu\n",
hmsg.mode, hmsg.id, hmsg.payload_len);
// Find the connection by ID.
DhtConnection *conn = nullptr;
for (int i = 0; i < DHT_MAX_CONNECTIONS; i++) {
if (!_connections[i].active) continue;
if (_connections[i].id == conn_id) {
conn = &_connections[i];
break;
}
}
if (!conn || conn->state < DHT_CONN_HOLEPUNCHING) {
// Not attributable to an attempt that is waiting on a relay. The id in
// the tag can be reused after a connection is freed, so a late reply
// must NOT be allowed to abort whatever holds that id now.
DHT_LOG("[DhtConnector] holepunch response: no matching conn for id=%u\n", conn_id);
return true;
}
// ⛔ AN UNREADABLE REPLY IS AN EXPLICIT ERROR, NOT A REASON TO WAIT.
//
// JS wires `c.relayClient.pair(...).on('error', onabort)` (connect.js:814):
// a relay error destroys the session immediately. These three branches used
// to log and return, leaving the attempt in HOLEPUNCHING until the 30-second
// wall clock — for a reply that will never become valid, because it is the
// answer to this request and it did not decode.
//
// docs/PARITY-hyperswarm.md §11.1: "explicit error | JS immediate | ours
// 30s". With SWARM_MAX_PARALLEL = 3 that is an outbound slot parked on a
// dead attempt for half a minute.
//
// The EMPTY reply above is deliberately NOT treated this way: "the relay
// could not reach the peer" is the case JS covers with its 15s relayTimeout,
// not with onabort, and failing it here would be more aggressive than the
// reference rather than equal to it.
if (!envelope_ok) {
_fail_connection(conn, "holepunch reply did not decode");
return true;
}
// Decrypt the payload using the holepunch secret.
std::vector<uint8_t> pt;
if (!dht_msg::secure_payload_decrypt(hmsg.payload, hmsg.payload_len,
conn->holepunch_secret, pt)) {
DHT_LOG("[DhtConnector] holepunch response: decrypt failed\n");
_fail_connection(conn, "holepunch reply failed to authenticate");
return true;
}
// Decode holepunch payload.
EncodeState ps = { 0, pt.size(), pt.data() };
dht_msg::HolepunchPayload payload;
if (!dht_msg::holepunch_payload_decode(ps, payload)) {
_fail_connection(conn, "holepunch payload did not decode");
return true;
}
DHT_LOG("[DhtConnector] holepunch response: fw=%d round=%d punching=%d connected=%d addrs=%d\n",
payload.firewall, payload.round, payload.punching, payload.connected,
payload.addresses_count);
_process_holepunch(conn, payload);
return true;
}
void DhtConnector::_process_holepunch(DhtConnection *conn,
const dht_msg::HolepunchPayload &payload) {
// Update remote info.
RemotePeerInfo rinfo;
memset(&rinfo, 0, sizeof(rinfo));
rinfo.firewall = payload.firewall;
rinfo.punching = payload.punching;
if (payload.addresses_count > 0) {
rinfo.address_count = payload.addresses_count;
for (int i = 0; i < payload.addresses_count; i++) {
rinfo.addresses[i] = payload.addresses[i];
}
}
conn->remote_firewall = payload.firewall;
conn->puncher.update_remote(rinfo);
// Start puncher once we have remote addresses from coordination.
if (!conn->puncher_started && rinfo.address_count > 0) {
conn->state = DHT_CONN_PUNCHING;
conn->puncher.start();
conn->puncher_started = true;
DHT_LOG("[DhtConnector] _process_holepunch: puncher started (%d remote addrs)\n",
rinfo.address_count);
}
// If remote reports connected, we're done.
if (payload.connected) {
// Remote says they're connected — we should be too.
}
// Send another holepunch message (advance the round).
if (conn->state == DHT_CONN_PUNCHING || conn->state == DHT_CONN_HOLEPUNCHING) {
_send_holepunch(conn);
}
}
const DhtConnection *DhtConnector::find_connection(uint32_t id) const {
for (int i = 0; i < DHT_MAX_CONNECTIONS; i++) {
if (_connections[i].active && _connections[i].id == id) return &_connections[i];
}
return nullptr;
}
void DhtConnector::on_find_peer_complete(uint32_t query_id) {
// The DHT has finished looking and, for any connection still in FINDING,
// found nobody — a peer that answered would have moved the connection on
// via on_peer_found. Waiting out DHT_CONNECT_TIMEOUT_MS for an answer we
// already have is what parked the slot for 30 seconds.
for (int i = 0; i < DHT_MAX_CONNECTIONS; i++) {
DhtConnection *conn = &_connections[i];
if (!conn->active) continue;
if (conn->state != DHT_CONN_FINDING) continue;
if (conn->find_peer_qid != query_id) continue;
conn->state = DHT_CONN_FAILED;
conn->error = "peer not found";
DHT_LOG("[DhtConnector] find_peer exhausted for conn %u — failing now "
"rather than at the timeout\n", conn->id);
if (on_failed) on_failed(conn->id, conn->remote_public_key, conn->error);
_free_connection(conn);
// _free_connection may compact, so do not advance i — the entry now at
// this index has not been examined. Same rule as the handshake sweep.
i--;
}
}
void DhtConnector::on_probe_received(const IPv4Address &from) {
// Route the probe to the puncher that's actually punching THIS peer.
//
// The previous code dispatched to the FIRST active puncher and broke, so a
// probe from peer A handed to peer B's puncher was dropped by B's own
// _is_remote_ip() check and A's puncher never saw it → A's holepunch never
// completed. With 2+ simultaneous punches (stale slot, retry, second peer)
// this is the intermittent two-peer mesh failure (#93). JS routes by socket
// ownership (each puncher owns its sockets); on our shared DHT socket the
// equivalent is to route by the remote IP each puncher already validates.
for (int i = 0; i < DHT_MAX_CONNECTIONS; i++) {
if (!_connections[i].active) continue;
if (_connections[i].state != DHT_CONN_PUNCHING &&
_connections[i].state != DHT_CONN_HOLEPUNCHING) continue;
if (!_connections[i].puncher.matches_remote(from)) continue;
_connections[i].puncher.on_probe_received(from);
return; // delivered to the owning puncher
}
// No puncher claims this source IP yet — a responder whose holepunch
// coordination (remote addresses) hasn't arrived. Echo via the first
// connection still in coordination so the exchange can bootstrap; the
// puncher echoes without connecting (it's not punching / no remote match).
for (int i = 0; i < DHT_MAX_CONNECTIONS; i++) {
if (!_connections[i].active) continue;
if (_connections[i].state != DHT_CONN_HOLEPUNCHING) continue;
_connections[i].puncher.on_probe_received(from);
return;
}
}
// ---- Finalize connection ----
void DhtConnector::_finalize_connection(DhtConnection *conn, const IPv4Address &addr) {
DHT_LOG("[DhtConnector] _finalize_connection: conn=%u addr=%d.%d.%d.%d:%d initiator=%d state=%d\n",
conn->id, addr.ip[0], addr.ip[1], addr.ip[2], addr.ip[3], addr.port,
conn->is_initiator, conn->state);
// ---- Connection deduplication ----
// Matches JS _handleServerConnection: if duplicate exists, compare keys to decide winner.
// Check ALL active connections to this peer (not just CONNECTED ones) to prevent
// two connections holepunching to the same address simultaneously (crashes libuv).
for (int di = 0; di < DHT_MAX_CONNECTIONS; di++) {
DhtConnection *existing = &_connections[di];
if (!existing->active || existing == conn) continue;
if (memcmp(existing->remote_public_key, conn->remote_public_key, 32) != 0) continue;
// JS: existingIsOutdated = existing.rawBytesRead > 0 && existing.rawBytesWritten > 0
bool existing_is_outdated = (existing->state == DHT_CONN_CONNECTED);
// JS: expectedInitiator = b4a.compare(conn.publicKey, conn.remotePublicKey) > 0
int cmp = memcmp(_dht->public_key(), conn->remote_public_key, 32);
bool expected_initiator = (cmp > 0);
// JS: keepNew = existingIsOutdated || (expectedInitiator === conn.isInitiator)
bool keep_new = existing_is_outdated || (expected_initiator == conn->is_initiator);
if (keep_new) {
DHT_LOG("[DhtConnector] _finalize: DEDUP — destroying existing conn %u (keep new %u)\n",
existing->id, conn->id);
if (on_failed) on_failed(existing->id, existing->remote_public_key, "duplicate connection (replaced)");
_free_connection(existing);
} else {
DHT_LOG("[DhtConnector] _finalize: DEDUP — destroying new conn %u (keep existing %u)\n",
conn->id, existing->id);
if (on_failed) on_failed(conn->id, conn->remote_public_key, "duplicate connection (kept existing)");
_free_connection(conn);
return;
}
break; // Only one duplicate possible after early dedup in on_handshake_request.
}
conn->state = DHT_CONN_CONNECTED;
conn->punched_address = addr;
conn->punched = true;
// Take winning birthday socket (if any) for UDX stream creation.
conn->punch_socket = conn->puncher.take_winning_socket();
// Track punch statistics.
_stats.total++;
HolepunchStage pstage = conn->puncher.stage();
if (pstage == HP_PUNCH_CONSISTENT) {
_stats.consistent_ok++;
} else if (pstage == HP_PUNCH_RANDOM || pstage == HP_PUNCH_BIRTHDAY) {
_stats.random_ok++;
if (_random_punch_count > 0) _random_punch_count--;
_last_random_punch = _now_ms;
} else if (pstage == HP_IDLE || pstage == HP_ANALYZE) {
// Direct connection (skipped punching).
_stats.open_ok++;
}
if (on_connected) {
on_connected(conn->id, *conn);
}
// Free the connection slot — UDX stream takes over with its own keepalive.
// The connection was copied (by value) into the on_connected callback,
// so we can safely release this slot for future connections.
_free_connection(conn);
}
// ---- Connection management ----
DhtConnection *DhtConnector::_alloc_connection(bool initiator, const uint8_t *remote_pk) {
for (int i = 0; i < DHT_MAX_CONNECTIONS; i++) {
if (!_connections[i].active) {
DhtConnection *conn = &_connections[i];
// Zero POD fields individually — do NOT memset the whole struct
// because Holepuncher contains std::function members.
conn->id = ++_conn_id_counter;
conn->state = DHT_CONN_IDLE;
conn->active = true;
conn->relay_since = 0;
conn->is_initiator = initiator;
memcpy(conn->remote_public_key, remote_pk, 32);
memset(&conn->relay_address, 0, sizeof(conn->relay_address));
conn->has_relay = false;
memset(&conn->server_address, 0, sizeof(conn->server_address));
conn->has_server_address = false;
conn->find_peer_qid = 0;
conn->handshake_id = 0;
memset(conn->holepunch_secret, 0, 32);
memset(conn->tx_key, 0, 32);
memset(conn->rx_key, 0, 32);
memset(conn->handshake_hash, 0, 64);
memset(&conn->remote_payload, 0, sizeof(conn->remote_payload));
conn->has_remote_payload = false;
conn->remote_stream_id = 0;
conn->remote_seq = 0;
// Full reset — stop() alone leaves _started/_connected stale,
// causing is_connected()=true on reused slots.
conn->puncher.reset();
conn->holepunch_round = 0;
memset(&conn->punched_address, 0, sizeof(conn->punched_address));
conn->punched = false;
conn->puncher_started = false;
conn->local_firewall = FIREWALL_UNKNOWN;
conn->remote_firewall = FIREWALL_UNKNOWN;
conn->started_at = 0;
conn->last_activity = 0;
conn->punch_socket = nullptr;
conn->error = nullptr;
return conn;
}
}
return nullptr;
}
DhtConnection *DhtConnector::_find_connection(uint32_t id) {
for (int i = 0; i < DHT_MAX_CONNECTIONS; i++) {
if (_connections[i].active && _connections[i].id == id) {
return &_connections[i];
}
}
return nullptr;
}
DhtConnection *DhtConnector::_find_connection_by_pk(const uint8_t *pk) {
for (int i = 0; i < DHT_MAX_CONNECTIONS; i++) {
if (_connections[i].active &&
memcmp(_connections[i].remote_public_key, pk, 32) == 0) {
return &_connections[i];
}
}
return nullptr;
}
void DhtConnector::_free_connection(DhtConnection *conn) {
conn->puncher.stop();
conn->active = false;
}
// ---- LAN detection helpers ----
void DhtConnector::_update_external_ip() {
const NatSampler &ns = _dht->rpc()->nat_sampler();
if (ns.address_count() > 0) {
// Use first (most common) external address.
_our_external_ip = ns.addresses()[0];
_has_external_ip = true;
}
}
bool DhtConnector::_is_same_lan(const DhtConnection *conn) const {
if (!conn->has_server_address) return false;
// Check loopback — always same machine.
static const uint8_t LOOPBACK[] = {127, 0, 0, 1};
if (memcmp(conn->server_address.ip, LOOPBACK, 4) == 0) return true;
// Check if server_address matches one of our local interfaces (same machine).
for (int i = 0; i < _local_address_count; i++) {
if (memcmp(_local_addresses[i].ip, conn->server_address.ip, 4) == 0) return true;
}
// Compare external IPs: if both peers have the same external IP,
// they're behind the same NAT → same LAN.
// JS: clientAddress.host === serverAddress.host
if (_has_external_ip) {
bool same_ext = (memcmp(_our_external_ip.ip, conn->server_address.ip, 4) == 0);
DHT_LOG("[DhtConnector] _is_same_lan: our_ext=%d.%d.%d.%d remote_ext=%d.%d.%d.%d → %s\n",
_our_external_ip.ip[0], _our_external_ip.ip[1],
_our_external_ip.ip[2], _our_external_ip.ip[3],
conn->server_address.ip[0], conn->server_address.ip[1],
conn->server_address.ip[2], conn->server_address.ip[3],
same_ext ? "YES" : "no");
if (same_ext) return true;
}
// Multi-NIC fallback: if the remote's advertised private addresses share
// a subnet with any of our local interfaces, they're on the same LAN.
// Covers the case where both peers are on 192.168.50.x but get different
// external IPs due to multiple gateways/VPN routes.
if (conn->has_remote_payload) {
for (int r = 0; r < conn->remote_payload.addresses4_count; r++) {
const IPv4Address &ra = conn->remote_payload.addresses4[r];
if (!_is_private_ip(ra.ip)) continue;
for (int l = 0; l < _local_address_count; l++) {
// /24 subnet match (same as _find_lan_address tier 0).
if (memcmp(_local_addresses[l].ip, ra.ip, 3) == 0) {
DHT_LOG("[DhtConnector] _is_same_lan: multi-NIC match — remote %d.%d.%d.%d local %d.%d.%d.%d\n",
ra.ip[0], ra.ip[1], ra.ip[2], ra.ip[3],
_local_addresses[l].ip[0], _local_addresses[l].ip[1],
_local_addresses[l].ip[2], _local_addresses[l].ip[3]);
return true;
}
}
}
}
return false;
}
bool DhtConnector::_is_private_ip(const uint8_t *ip) {
// RFC 1918 private ranges.
if (ip[0] == 10) return true; // 10.0.0.0/8
if (ip[0] == 172 && ip[1] >= 16 && ip[1] <= 31) return true; // 172.16.0.0/12
if (ip[0] == 192 && ip[1] == 168) return true; // 192.168.0.0/16
if (ip[0] == 127) return true; // loopback
return false;
}
bool DhtConnector::_find_lan_address(const DhtConnection *conn, IPv4Address &out) const {
if (!conn->has_remote_payload || conn->remote_payload.addresses4_count == 0)
return false;
// Try to find a remote address on the same subnet as one of our interfaces.
// Three tiers: /24 exact → /16 → /8 (matches JS matchAddress from holepuncher.js).
for (int tier = 0; tier < 3; tier++) {
int mask_bytes = 3 - tier; // 3=match first 3 bytes(/24), 2=(/16), 1=(/8)
for (int r = 0; r < conn->remote_payload.addresses4_count; r++) {
const IPv4Address &ra = conn->remote_payload.addresses4[r];
// Skip public IPs — only match private addresses.
if (!_is_private_ip(ra.ip)) continue;
for (int l = 0; l < _local_address_count; l++) {
if (memcmp(_local_addresses[l].ip, ra.ip, mask_bytes) == 0) {
out = ra;
DHT_LOG("[DhtConnector] _find_lan_address: matched remote %d.%d.%d.%d:%d "
"with local %d.%d.%d.%d (tier=%d)\n",
ra.ip[0], ra.ip[1], ra.ip[2], ra.ip[3], ra.port,
_local_addresses[l].ip[0], _local_addresses[l].ip[1],
_local_addresses[l].ip[2], _local_addresses[l].ip[3], tier);
return true;
}
}
}
}
// Fallback: return first private address from remote (any subnet).
for (int r = 0; r < conn->remote_payload.addresses4_count; r++) {
const IPv4Address &ra = conn->remote_payload.addresses4[r];
if (_is_private_ip(ra.ip)) {
out = ra;
return true;
}
}
return false;
}
// ---- Encode local payload ----
// ---- Diagnostics ----
int DhtConnector::get_connections_info(DhtConnectionInfo *out, int max_count, uint64_t now_ms) const {
int count = 0;
for (int i = 0; i < DHT_MAX_CONNECTIONS && count < max_count; i++) {
const DhtConnection &c = _connections[i];
if (!c.active) continue;
DhtConnectionInfo &info = out[count];
info.id = c.id;
info.state = c.state;
info.is_initiator = c.is_initiator;
memcpy(info.remote_public_key, c.remote_public_key, 32);
info.relay_address = c.relay_address;
info.has_relay = c.has_relay;
info.server_address = c.server_address;
info.has_server_address = c.has_server_address;
info.punched_address = c.punched_address;
info.punched = c.punched;
info.holepunch_round = c.holepunch_round;
info.local_firewall = c.local_firewall;
info.remote_firewall = c.remote_firewall;
info.started_at = c.started_at;
info.elapsed_ms = (now_ms > c.started_at) ? (now_ms - c.started_at) : 0;
info.error = c.error;
// Copy remote addresses from NoisePayload if available.
if (c.has_remote_payload) {
info.remote_address_count = c.remote_payload.addresses4_count;
for (int j = 0; j < c.remote_payload.addresses4_count && j < 8; j++) {
info.remote_addresses[j] = c.remote_payload.addresses4[j];
}
} else {
info.remote_address_count = 0;
}
count++;
}
return count;
}