#include "udx_transport.h"
#include "debug_log.h"
#include <cstdio>
#include <cstdlib>
#include <cstring>
// Header byte prepended to every packet:
// [1 byte: mode | channel<<2] [payload...]
static const int HEADER_SIZE = 1;
static uint8_t encode_header(PacketTransferMode mode, int32_t channel) {
return static_cast<uint8_t>((mode & 0x3) | ((channel & 0x3F) << 2));
}
static void decode_header(uint8_t h, PacketTransferMode &mode, int32_t &channel) {
mode = static_cast<PacketTransferMode>(h & 0x3);
channel = (h >> 2) & 0x3F;
}
// ---- Length-prefixed framing (matches JS SecretStream uint24 LE) ----
// UDX reliable streams are byte streams — writes can coalesce into one read.
// Frame format: [3-byte uint24 little-endian length][payload]
static const int FRAME_PREFIX_SIZE = 3;
static void write_frame_length(uint8_t *buf, uint32_t len) {
buf[0] = len & 0xFF;
buf[1] = (len >> 8) & 0xFF;
buf[2] = (len >> 16) & 0xFF;
}
static uint32_t read_frame_length(const uint8_t *buf) {
return static_cast<uint32_t>(buf[0])
| (static_cast<uint32_t>(buf[1]) << 8)
| (static_cast<uint32_t>(buf[2]) << 16);
}
// ---- SecretStream streamId derivation (matches JS @hyperswarm/secret-stream) ----
// JS: crypto.namespace('hyperswarm/secret-stream', 3) produces NS_INITIATOR (counter=0)
// and NS_RESPONDER (counter=1). Then: streamId = BLAKE2b(NS_KEY, key=handshakeHash).
// We precompute the namespace keys since the input string is fixed.
static void compute_stream_id(uint8_t *out, const uint8_t *handshake_hash, bool is_initiator) {
// Step 1: ns[0:32] = BLAKE2b-256("hyperswarm/secret-stream")
uint8_t ns[33];
crypto_generichash(ns, 32,
(const uint8_t *)"hyperswarm/secret-stream", 24,
nullptr, 0);
// Step 2: append counter byte (0 = initiator, 1 = responder)
ns[32] = is_initiator ? 0 : 1;
// Step 3: ns_key = BLAKE2b-256(ns[0:33], no key)
uint8_t ns_key[32];
crypto_generichash(ns_key, 32, ns, 33, nullptr, 0);
// Step 4: streamId = BLAKE2b-256(message=ns_key, key=handshake_hash)
crypto_generichash(out, 32, ns_key, 32, handshake_hash, 64);
}
// ---- Construction / Destruction ----
UdxTransport::UdxTransport()
: _running(false), _is_server(false), _bound_port(0),
_next_stream_id(1), _peer_id_counter(1),
_encrypted(false),
_dht_active(false), _dht_bootstrapped(false), _dht_rpc(nullptr), _hyperdht(nullptr),
_dht_connector(nullptr),
_swarm_manager(nullptr) {
memset(&_loop, 0, sizeof(_loop));
memset(&_udx, 0, sizeof(_udx));
memset(&_socket, 0, sizeof(_socket));
memset(&_async_handle, 0, sizeof(_async_handle));
memset(&_tick_timer, 0, sizeof(_tick_timer));
memset(&_local_keypair, 0, sizeof(_local_keypair));
memset(_remote_public_key, 0, 32);
memset(&_dht_keypair, 0, sizeof(_dht_keypair));
}
UdxTransport::~UdxTransport() {
stop();
}
// ---- Public API (called from Godot main thread) ----
bool UdxTransport::start_server(int port) {
if (_running.load()) return false;
_is_server.store(true);
_bound_port = port;
_running.store(true);
_thread = std::thread(&UdxTransport::_thread_func, this);
return true;
}
bool UdxTransport::start_client(const char *address, int port) {
if (_running.load()) return false;
_is_server.store(false);
// Queue a connect command — the thread will process it after starting.
Command cmd;
cmd.type = CMD_CONNECT_TO;
cmd.address = address;
cmd.port = port;
cmd.peer_id = 1; // Server is always peer 1
_queue_command(std::move(cmd));
_running.store(true);
_thread = std::thread(&UdxTransport::_thread_func, this);
return true;
}
bool UdxTransport::start_server_encrypted(int port, const NoiseKeypair &keypair) {
if (_running.load()) return false;
_encrypted = true;
memcpy(&_local_keypair, &keypair, sizeof(NoiseKeypair));
_is_server.store(true);
_bound_port = port;
_running.store(true);
_thread = std::thread(&UdxTransport::_thread_func, this);
return true;
}
bool UdxTransport::start_client_encrypted(const char *address, int port,
const uint8_t *remote_public_key) {
if (_running.load()) return false;
_encrypted = true;
memcpy(_remote_public_key, remote_public_key, 32);
// Generate an ephemeral keypair for this client connection.
_local_keypair = NoiseHandshake::generate_keypair();
_is_server.store(false);
Command cmd;
cmd.type = CMD_CONNECT_TO;
cmd.address = address;
cmd.port = port;
cmd.peer_id = 1;
memcpy(cmd.remote_public_key, remote_public_key, 32);
_queue_command(std::move(cmd));
_running.store(true);
_thread = std::thread(&UdxTransport::_thread_func, this);
return true;
}
void UdxTransport::stop() {
if (!_running.load()) return;
_running.store(false);
// Wake up the event loop so it can exit.
_wake();
if (_thread.joinable()) {
_thread.join();
}
// Cleanup SwarmManager (before DHT — it references DHT objects).
delete _swarm_manager;
_swarm_manager = nullptr;
// Cleanup DHT (after thread join — these are UDX-thread-only objects).
delete _dht_connector;
_dht_connector = nullptr;
delete _hyperdht;
_hyperdht = nullptr;
delete _dht_rpc;
_dht_rpc = nullptr;
_dht_active.store(false);
_dht_bootstrapped.store(false);
// Cleanup peers.
for (auto &pair : _peers) {
delete pair.second;
}
_peers.clear();
_peers_by_pubkey.clear();
}
void UdxTransport::send(int32_t target_peer, const uint8_t *data, int32_t size,
PacketTransferMode mode, int32_t channel) {
Command cmd;
cmd.type = CMD_SEND;
cmd.peer_id = target_peer;
cmd.data.resize(HEADER_SIZE + size);
cmd.data[0] = encode_header(mode, channel);
if (size > 0) {
memcpy(cmd.data.data() + HEADER_SIZE, data, size);
}
cmd.mode = mode;
cmd.channel = channel;
_queue_command(std::move(cmd));
_wake();
}
void UdxTransport::disconnect_peer(int32_t peer_id, bool force) {
Command cmd;
cmd.type = CMD_DISCONNECT;
cmd.peer_id = peer_id;
cmd.force = force;
_queue_command(std::move(cmd));
_wake();
}
bool UdxTransport::pop_peer_event(PeerEvent &out) {
std::lock_guard<std::mutex> lock(_events_mutex);
if (_peer_events.empty()) return false;
out = _peer_events.front();
_peer_events.pop_front();
return true;
}
bool UdxTransport::pop_peer_found_event(PeerFoundEvent &out) {
std::lock_guard<std::mutex> lock(_events_mutex);
if (_peer_found_events.empty()) return false;
out = _peer_found_events.front();
_peer_found_events.pop_front();
return true;
}
bool UdxTransport::pop_storage_event(DhtStorageEvent &out) {
std::lock_guard<std::mutex> lock(_events_mutex);
if (_storage_events.empty()) return false;
out = std::move(_storage_events.front());
_storage_events.pop_front();
return true;
}
// ---- DHT Public API (called from Godot main thread) ----
const char *const UdxTransport::DEFAULT_BOOTSTRAP[3] = {
"88.99.3.86:49737",
"142.93.90.113:49737",
"138.68.147.8:49737",
};
void UdxTransport::start_dht(const Ed25519KeyPair &kp,
const char **bootstrap_nodes, int bootstrap_count) {
memcpy(&_dht_keypair, &kp, sizeof(Ed25519KeyPair));
// COPIED, NOT BORROWED. This function does not use the list — it queues
// CMD_DHT_INIT and returns, and _init_dht parses it later on the UDX thread.
// Keeping the caller's pointers made every peer's bootstrap argument live
// across that thread boundary for an unbounded time, which nothing in
// join_dht's contract says it does; the caller kept them alive in statics
// SHARED BY EVERY INSTANCE, so a second peer's join rewrote the list the
// first peer's transport was still pointing at. Copying here is what makes
// the contract "the list is read during this call" true.
//
// Written before _queue_command, so the queue mutex publishes it: the UDX
// thread acquires the same lock to pop the command it is reading this for.
_bootstrap.clear();
_bootstrap.reserve(bootstrap_count > 0 ? (size_t)bootstrap_count : 0);
for (int i = 0; i < bootstrap_count; i++) {
if (bootstrap_nodes && bootstrap_nodes[i]) _bootstrap.emplace_back(bootstrap_nodes[i]);
}
// No usable entry — including the {nullptr} that "omit --bootstrap" produced.
// Falling back HERE rather than in set_bootstrap is deliberate: this reacts to
// "the caller named no nodes", while set_bootstrap sees "no node PARSED", and
// silently substituting the public DHT for three unreachable addresses would
// hide a typo instead of the omission it is meant to cover.
if (_bootstrap.empty()) {
for (const char *node : DEFAULT_BOOTSTRAP) _bootstrap.emplace_back(node);
}
Command cmd;
cmd.type = CMD_DHT_INIT;
_queue_command(std::move(cmd));
if (!_running.load()) {
// Start the transport thread (binds to random port).
_running.store(true);
_thread = std::thread(&UdxTransport::_thread_func, this);
} else {
_wake();
}
}
void UdxTransport::dht_connect(const uint8_t *remote_public_key) {
Command cmd;
cmd.type = CMD_DHT_CONNECT;
memcpy(cmd.remote_public_key, remote_public_key, 32);
_queue_command(std::move(cmd));
_wake();
}
void UdxTransport::dht_listen() {
Command cmd;
cmd.type = CMD_DHT_LISTEN;
_queue_command(std::move(cmd));
_wake();
}
void UdxTransport::dht_announce(const uint8_t *topic) {
Command cmd;
cmd.type = CMD_DHT_ANNOUNCE;
cmd.data.assign(topic, topic + 32);
_queue_command(std::move(cmd));
_wake();
}
void UdxTransport::dht_lookup(const uint8_t *topic) {
Command cmd;
cmd.type = CMD_DHT_LOOKUP;
cmd.data.assign(topic, topic + 32);
_queue_command(std::move(cmd));
_wake();
}
void UdxTransport::dht_join_topic(const uint8_t *topic, bool server, bool client) {
Command cmd;
cmd.type = CMD_DHT_JOIN_TOPIC;
cmd.data.assign(topic, topic + 32);
cmd.server = server;
cmd.client = client;
_queue_command(std::move(cmd));
_wake();
}
void UdxTransport::dht_leave_topic(const uint8_t *topic) {
Command cmd;
cmd.type = CMD_DHT_LEAVE_TOPIC;
cmd.data.assign(topic, topic + 32);
_queue_command(std::move(cmd));
_wake();
}
void UdxTransport::dht_immutable_put(const uint8_t *value, size_t value_len) {
Command cmd;
cmd.type = CMD_DHT_IMMUTABLE_PUT;
cmd.data.assign(value, value + value_len);
_queue_command(std::move(cmd));
_wake();
}
void UdxTransport::dht_immutable_get(const uint8_t *key) {
Command cmd;
cmd.type = CMD_DHT_IMMUTABLE_GET;
cmd.data.assign(key, key + 32);
_queue_command(std::move(cmd));
_wake();
}
void UdxTransport::dht_mutable_put(const uint8_t *value, size_t value_len, uint64_t seq) {
Command cmd;
cmd.type = CMD_DHT_MUTABLE_PUT;
cmd.data.assign(value, value + value_len);
cmd.seq = seq;
_queue_command(std::move(cmd));
_wake();
}
void UdxTransport::dht_mutable_get(const uint8_t *public_key, uint64_t min_seq) {
Command cmd;
cmd.type = CMD_DHT_MUTABLE_GET;
cmd.data.assign(public_key, public_key + 32);
cmd.seq = min_seq;
_queue_command(std::move(cmd));
_wake();
}
// ---- Background Thread ----
static void _async_cb(uv_async_t *handle) {
// Just wakes up the event loop — commands processed in timer tick.
(void)handle;
}
static void _tick_cb(uv_timer_t *handle) {
UdxTransport *transport = static_cast<UdxTransport *>(handle->data);
transport->_process_commands();
transport->_sweep_keepalive(uv_now(handle->loop));
}
// ⛔ THE ONLY WAY TO WAKE THE LOOP, because the guard was previously written at
// exactly ONE of twenty call sites.
//
// `_async_handle` is a uv_async_t that does not exist until `_thread_func` has
// run `uv_async_init` on it. Signalling it before that reads a null callback
// pointer and jumps to address zero. run_on_loop checked `_async_ready`;
// swarm_join, swarm_leave, the peer variants, every dht_* command, send,
// disconnect_peer, start_dht and stop did not — nineteen unguarded senders and
// one guarded one.
//
// It was reachable the moment a client issued a command promptly after start:
// measured as a SEGV in 7 runs of 20 of swarmd's test_client, with the UDX
// thread still inside udx_socket_bind and thread 4 at `0x0 in ?? ()`. Nothing
// before it called anything that fast — the e2e gates all wait for dht=up
// first, which is why a race this plain survived.
//
// Commands are queued regardless, and `_thread_func` sends to itself the
// instant the handle exists, so a wake dropped here is a wake deferred by
// microseconds and not a command lost.
void UdxTransport::_wake() {
if (_async_ready) uv_async_send(&_async_handle);
}
void UdxTransport::_thread_func() {
uv_loop_init(&_loop);
udx_init(&_loop, &_udx, nullptr);
// Async handle for waking up the loop from main thread.
uv_async_init(&_loop, &_async_handle, _async_cb);
// Only now may anyone signal it. Publish AFTER init, and flush whatever was
// queued while the thread was starting.
_async_ready = true;
_wake();
// Periodic timer for processing commands.
uv_timer_init(&_loop, &_tick_timer);
_tick_timer.data = this;
uv_timer_start(&_tick_timer, _tick_cb, 0, 1); // 1ms tick
// Initialize socket.
udx_socket_init(&_udx, &_socket, nullptr);
_socket.data = this;
struct sockaddr_in bind_addr;
uv_ip4_addr("0.0.0.0", _is_server.load() ? _bound_port : 0, &bind_addr);
udx_socket_bind(&_socket, (struct sockaddr *)&bind_addr, 0);
// Get actual bound port.
struct sockaddr_in local_addr;
int addr_len = sizeof(local_addr);
udx_socket_getsockname(&_socket, (struct sockaddr *)&local_addr, &addr_len);
_bound_port = ntohs(local_addr.sin_port);
// Always enable socket recv — needed for DHT messages, holepunch probes,
// and server-side peer discovery.
udx_socket_recv_start(&_socket, _on_socket_recv);
// Run the event loop until stopped.
while (_running.load()) {
uv_run(&_loop, UV_RUN_NOWAIT);
// Small sleep to avoid busy-waiting.
std::this_thread::sleep_for(std::chrono::microseconds(500));
}
// Cleanup: destroy all streams.
for (auto &pair : _peers) {
if (pair.second->connected) {
udx_stream_destroy(&pair.second->stream);
}
}
// Close handles.
uv_timer_stop(&_tick_timer);
uv_close((uv_handle_t *)&_tick_timer, nullptr);
uv_close((uv_handle_t *)&_async_handle, nullptr);
udx_socket_close(&_socket);
// Run loop to process close callbacks.
uv_run(&_loop, UV_RUN_DEFAULT);
uv_loop_close(&_loop);
}
static UdxTransport::StreamInfo stream_info_of (const PeerConnection *peer) {
UdxTransport::StreamInfo info;
info.peer_id = peer->godot_peer_id;
info.is_initiator = peer->is_noise_initiator;
memcpy(info.handshake_hash, peer->_handshake_hash, 64);
memcpy(info.remote_public_key, peer->remote_public_key, 32);
return info;
}
void UdxTransport::set_stream_consumer(std::function<void(const StreamInfo &)> open,
std::function<void(int32_t, const uint8_t *, size_t)> frame,
std::function<void(int32_t)> drain,
std::function<void(int32_t)> close) {
run_on_loop([this, open = std::move(open), frame = std::move(frame), drain = std::move(drain),
close = std::move(close)]() mutable {
_stream_open = std::move(open);
_stream_frame = std::move(frame);
_stream_drain = std::move(drain);
_stream_close = std::move(close);
if (!_stream_open) return;
// CATCH UP, as set_mux_callbacks does: a stream that is already up and never
// got the internal mux becomes the consumer's.
for (auto &kv : _peers) {
PeerConnection *p = kv.second;
if (!p || !p->noise_ready || p->mux || p->stream_consumer) continue;
p->stream_consumer = true;
_stream_open(stream_info_of(p));
}
});
}
bool UdxTransport::send_frame(int32_t peer_id, const uint8_t *data, size_t len) {
PeerConnection *peer = _get_peer(peer_id);
if (!peer || !peer->connected || !peer->noise_ready || !peer->stream_consumer) return false;
_send_encrypted(peer, data, len);
return peer->writable;
}
void UdxTransport::close_stream(int32_t peer_id) {
PeerConnection *peer = _get_peer(peer_id);
if (!peer || !peer->connected) return;
udx_stream_destroy(&peer->stream); // _on_stream_close follows, which tells the consumer
}
void UdxTransport::set_mux_callbacks(std::function<void(int32_t, Protomux *)> ready,
std::function<void(int32_t)> gone) {
run_on_loop([this, ready = std::move(ready), gone = std::move(gone)]() mutable {
on_mux_ready = std::move(ready);
on_mux_gone = std::move(gone);
if (!on_mux_ready) return;
// CATCH UP. Every peer that meshed before this consumer registered gets the
// announcement it would have received had it been here — otherwise "attach
// after the swarm is up", which is the ordering every caller actually uses,
// silently yields a consumer with no peers.
for (auto &kv : _peers) {
PeerConnection *p = kv.second;
if (p && p->mux && p->mux_ready) on_mux_ready(p->godot_peer_id, p->mux);
}
});
}
void UdxTransport::run_on_loop(std::function<void()> task) {
{
std::lock_guard<std::mutex> lock(_tasks_mutex);
_tasks.push_back(std::move(task));
}
// Queue always; wake only once the handle exists. See _wake().
_wake();
}
void UdxTransport::_process_commands() {
// Consumer tasks first: they are the ones that were told "this will run on the
// loop", and running them before the transport's own commands keeps that
// promise ordered against anything they queue in turn.
{
std::deque<std::function<void()>> tasks;
{
std::lock_guard<std::mutex> lock(_tasks_mutex);
tasks.swap(_tasks);
}
for (auto &t : tasks) t();
}
std::deque<Command> cmds;
{
std::lock_guard<std::mutex> lock(_cmd_mutex);
cmds.swap(_commands);
}
for (auto &cmd : cmds) {
switch (cmd.type) {
case CMD_CONNECT_TO: {
// Client connecting to server.
struct sockaddr_in addr;
uv_ip4_addr(cmd.address.c_str(), cmd.port, &addr);
uint32_t local_sid = _next_stream_id++;
uint32_t remote_sid = 1; // Server's first stream
PeerConnection *peer = _add_peer(cmd.peer_id, local_sid, remote_sid,
(struct sockaddr *)&addr);
if (peer) {
peer->connected = true;
if (_encrypted) {
// Start Noise handshake as initiator.
peer->encrypted = true;
peer->is_noise_initiator = true;
peer->handshake = new NoiseHandshake(true, _local_keypair);
peer->noise = new NoiseStream();
// Send first handshake message (msg 1: ephemeral key).
auto msg1 = peer->handshake->send();
std::vector<uint8_t> framed(1 + msg1.size());
framed[0] = NOISE_MSG_HANDSHAKE;
memcpy(framed.data() + 1, msg1.data(), msg1.size());
_send_raw(peer, framed.data(), framed.size());
// Don't emit connected yet — wait for handshake completion.
} else {
// Unencrypted — immediately connected.
std::lock_guard<std::mutex> lock(_events_mutex);
_peer_events.push_back({cmd.peer_id, true});
}
}
break;
}
case CMD_SEND: {
auto send_to_peer = [&](PeerConnection *peer) {
if (!peer || !peer->connected) {
return;
}
// If encrypted but not yet ready, drop the packet.
if (peer->encrypted && !peer->noise_ready) {
return;
}
// Protomux path: send through godot-rpc channel (type 0).
// Strip the 1-byte header — protomux handles framing.
if (peer->mux) {
auto &channels = peer->mux->local_channels();
if (!channels.empty() && channels[0] &&
channels[0]->opened && !channels[0]->closed) {
const uint8_t *game_data = cmd.data.data() + HEADER_SIZE;
int32_t game_len = static_cast<int32_t>(cmd.data.size()) - HEADER_SIZE;
if (game_len > 0) {
channels[0]->send(0, game_data, game_len);
}
}
return;
}
// Legacy path (non-protomux peers).
const uint8_t *payload = cmd.data.data();
int32_t payload_size = static_cast<int32_t>(cmd.data.size());
std::vector<uint8_t> encrypted_buf;
if (peer->encrypted && peer->noise_ready) {
// Encrypt the full payload (header + data).
encrypted_buf = peer->noise->encrypt(payload, payload_size);
payload = encrypted_buf.data();
payload_size = static_cast<int32_t>(encrypted_buf.size());
}
if (cmd.mode == PACKET_RELIABLE) {
// Length-prefixed framing for reliable byte stream.
size_t framed_len = FRAME_PREFIX_SIZE + payload_size;
uint8_t *copy = (uint8_t *)malloc(framed_len);
write_frame_length(copy, static_cast<uint16_t>(payload_size));
memcpy(copy + FRAME_PREFIX_SIZE, payload, payload_size);
uv_buf_t buf = uv_buf_init((char *)copy, static_cast<unsigned int>(framed_len));
udx_stream_write_t *req = (udx_stream_write_t *)malloc(
udx_stream_write_sizeof(1));
req->data = copy;
udx_stream_write(req, &peer->stream, &buf, 1, _on_write_ack);
} else {
// Unreliable send — message-based, no framing needed.
uint8_t *copy = (uint8_t *)malloc(payload_size);
memcpy(copy, payload, payload_size);
uv_buf_t buf = uv_buf_init((char *)copy, payload_size);
udx_stream_send_t *req = (udx_stream_send_t *)malloc(
sizeof(udx_stream_send_t));
req->data = copy;
udx_stream_send(req, &peer->stream, &buf, 1, _on_send_complete);
}
};
if (cmd.peer_id == 0) {
// Broadcast.
for (auto &pair : _peers) {
send_to_peer(pair.second);
}
} else if (cmd.peer_id < 0) {
// Send to all except abs(peer_id).
int32_t exclude = -cmd.peer_id;
for (auto &pair : _peers) {
if (pair.first != exclude) {
send_to_peer(pair.second);
}
}
} else {
// Unicast.
PeerConnection *peer = _get_peer(cmd.peer_id);
send_to_peer(peer);
}
break;
}
case CMD_DISCONNECT: {
PeerConnection *peer = _get_peer(cmd.peer_id);
if (peer && peer->connected) {
udx_stream_destroy(&peer->stream);
peer->connected = false;
{
std::lock_guard<std::mutex> lock(_events_mutex);
_peer_events.push_back({cmd.peer_id, false});
}
_remove_peer(cmd.peer_id);
}
break;
}
case CMD_DHT_INIT: {
_init_dht();
break;
}
case CMD_DHT_CONNECT: {
if (_dht_connector) {
_dht_connector->connect(cmd.remote_public_key);
}
break;
}
case CMD_DHT_LISTEN: {
if (_dht_connector) {
_dht_connector->listen();
}
break;
}
case CMD_DHT_ANNOUNCE: {
if (_hyperdht && cmd.data.size() == 32) {
_hyperdht->announce(cmd.data.data());
}
break;
}
case CMD_DHT_LOOKUP: {
if (_hyperdht && cmd.data.size() == 32) {
_hyperdht->lookup(cmd.data.data());
}
break;
}
case CMD_DHT_JOIN_TOPIC: {
if (_hyperdht && cmd.data.size() == 32) {
_hyperdht->join_topic(cmd.data.data(), cmd.server, cmd.client);
}
break;
}
case CMD_DHT_LEAVE_TOPIC: {
if (_hyperdht && cmd.data.size() == 32) {
_hyperdht->leave_topic(cmd.data.data());
}
break;
}
case CMD_DHT_IMMUTABLE_PUT: {
if (_hyperdht && !cmd.data.empty()) {
_hyperdht->immutable_put(cmd.data.data(), cmd.data.size());
}
break;
}
case CMD_DHT_IMMUTABLE_GET: {
if (_hyperdht && cmd.data.size() == 32) {
_hyperdht->immutable_get(cmd.data.data());
}
break;
}
case CMD_DHT_MUTABLE_PUT: {
if (_hyperdht && !cmd.data.empty()) {
_hyperdht->mutable_put(cmd.data.data(), cmd.data.size(), cmd.seq);
}
break;
}
case CMD_DHT_MUTABLE_GET: {
if (_hyperdht && cmd.data.size() == 32) {
_hyperdht->mutable_get(cmd.data.data(), cmd.seq);
}
break;
}
case CMD_SWARM_JOIN: {
if (_swarm_manager && cmd.data.size() == 32) {
_swarm_manager->join(cmd.data.data(), cmd.server, cmd.client);
// Also drive the underlying DHT operations.
if (cmd.server && _dht_connector) _dht_connector->listen();
if (_hyperdht) _hyperdht->join_topic(cmd.data.data(), cmd.server, cmd.client);
}
break;
}
case CMD_SWARM_LEAVE: {
if (_swarm_manager && cmd.data.size() == 32) {
_swarm_manager->leave(cmd.data.data());
if (_hyperdht) _hyperdht->leave_topic(cmd.data.data());
}
break;
}
case CMD_SWARM_JOIN_PEER: {
if (_swarm_manager) {
_swarm_manager->join_peer(cmd.remote_public_key);
}
break;
}
case CMD_SWARM_LEAVE_PEER: {
if (_swarm_manager) {
_swarm_manager->leave_peer(cmd.remote_public_key);
}
break;
}
}
}
// Tick DHT objects on every timer cycle.
if (_dht_rpc) {
uint64_t now = uv_now(&_loop);
_dht_rpc->tick(now);
if (_hyperdht) _hyperdht->tick(now);
if (_dht_connector) _dht_connector->tick(now);
// Tick SwarmManager and drain events to main thread.
if (_swarm_manager) {
_swarm_manager->tick(now);
SwarmEvent evt;
while (_swarm_manager->pop_event(evt)) {
std::lock_guard<std::mutex> lock(_events_mutex);
_swarm_events.push_back(evt);
}
}
}
}
// ---- libudx Callbacks (static, run on UDX thread) ----
void UdxTransport::_dispatch_incoming(PeerConnection *peer,
const uint8_t *buf_data, ssize_t buf_len,
PacketTransferMode mode_override) {
if (buf_len < 1) return;
uint8_t first_byte = buf_data[0];
// Route based on connection state, not byte prefix.
// Matches JS secret-stream: _setup flag determines routing.
// After noise_ready, ALL data is encrypted data — never check byte prefix.
if (peer->encrypted && peer->noise_ready) {
// Setup complete — decrypt data. No byte-prefix check (matches JS _setup=false).
std::vector<uint8_t> pt;
if (!peer->noise->decrypt(buf_data, static_cast<size_t>(buf_len), pt)) {
return; // Auth failure — drop.
}
// ⛔ ANY AUTHENTICATED FRAME IS EVIDENCE THIS PEER IS ALIVE, whatever it
// carries. That is the signal _sweep_keepalive waits on, and it is
// stamped here rather than in the protomux handler so a frame for a
// channel nobody opened still counts.
peer->last_recv_ms = _loop_now_ms;
if (pt.empty()) {
// The keepalive. @hyperswarm/secret-stream eats these on receipt
// ("assume the other side has it enabled also") and so do we —
// forwarding an empty frame to protomux would decode a channel id of
// 0 and a type of 0 out of nothing at all.
peer->peer_keepalive = true;
return;
}
// The stream consumer's connection: the message, as is.
if (peer->stream_consumer) {
if (_stream_frame) _stream_frame(peer->godot_peer_id, pt.data(), pt.size());
return;
}
// Route through protomux if available.
if (peer->mux) {
DHT_LOG("[UdxTransport] dispatch: routing %zu decrypted bytes to protomux for peer %d\n",
pt.size(), peer->godot_peer_id);
peer->mux->recv(pt.data(), pt.size());
return;
}
// Legacy fallback: 1-byte header path (pre-protomux peers).
if (pt.size() < (size_t)HEADER_SIZE) return;
PacketTransferMode mode;
int32_t channel;
decode_header(pt[0], mode, channel);
mode = mode_override; // Use the actual transport mode.
Packet pkt(pt.data() + HEADER_SIZE,
static_cast<int32_t>(pt.size()) - HEADER_SIZE,
peer->godot_peer_id, channel, mode);
_incoming.push(std::move(pkt));
} else if (peer->encrypted) {
// Encrypted setup phase (noise_ready=false). Route by connection state.
if (peer->handshake != nullptr) {
// Still in direct-path Noise handshake — messages use 0xFE prefix.
if (first_byte == NOISE_MSG_HANDSHAKE) {
_handle_noise_message(
peer, NOISE_MSG_HANDSHAKE,
buf_data + 1, static_cast<size_t>(buf_len - 1));
}
} else if (!peer->header_received) {
// Post-handshake, waiting for header: [32B streamId][24B header] (no type prefix).
_handle_noise_message(
peer, NOISE_MSG_HEADER,
buf_data, static_cast<size_t>(buf_len));
}
// Data before noise_ready: drop silently (can't decrypt yet).
} else {
// Unencrypted path.
if (buf_len < HEADER_SIZE) return;
PacketTransferMode mode;
int32_t channel;
decode_header(buf_data[0], mode, channel);
mode = mode_override;
const uint8_t *payload_start = buf_data + HEADER_SIZE;
int32_t payload_size = static_cast<int32_t>(buf_len) - HEADER_SIZE;
Packet pkt(payload_start, payload_size, peer->godot_peer_id, channel, mode);
_incoming.push(std::move(pkt));
}
}
void UdxTransport::_on_stream_read(udx_stream_t *stream, ssize_t read_len, const uv_buf_t *buf) {
if (read_len <= 0) return;
PeerConnection *peer = static_cast<PeerConnection *>(stream->data);
if (!peer) return;
UdxTransport *transport = static_cast<UdxTransport *>(stream->socket->data);
if (!transport) {
fprintf(stderr, "[UdxTransport] stream_read: DROPPING %zd bytes — socket->data is NULL!\n", read_len);
return;
}
// Log first read per peer to confirm data flow.
if (!peer->noise_ready && !peer->header_received) {
DHT_LOG("[UdxTransport] stream_read: first data from peer %d (%zd bytes)\n",
peer->godot_peer_id, read_len);
}
// Append to per-peer receive buffer (handles coalesced reads).
peer->recv_buf.insert(peer->recv_buf.end(),
(const uint8_t *)buf->base, (const uint8_t *)buf->base + read_len);
// THE BOUND ON THIS BUFFER IS STRUCTURAL, NOT A CONSTANT WE PICK.
//
// There used to be a flat `if (recv_buf.size() > 262144) disconnect` here, and
// it was wrong in a way that only a real asset showed. What accumulates in
// recv_buf between drains is, by definition, ONE PARTIAL FRAME — so capping
// the buffer caps the largest frame this transport can ever reassemble. At
// 256 KB that is 14x smaller than what the protocol riding on top is entitled
// to send: liboverlay pages chunkHashes at MAX_CHUNKS_PER_MSG = 100000
// entries (~3.7 MB) and the JS reference also allows a 4 MiB tree page, both
// sized against NOISE_MAX_ATOMIC_WRITE. A conformant holder sending a legal
// page therefore killed the connection, and the requester saw only "every
// holder disconnected" — see hyper-games #18 and layer 13.
//
// The real ceiling is the frame prefix itself: uint24 LE tops out at
// 16777215, which IS NOISE_MAX_ATOMIC_WRITE (16 MiB - 1) exactly. A peer
// cannot ask us to buffer more than that because it cannot express more than
// that, so worst-case residency is one max frame per peer — the same bound
// the JS side accepts over secret-stream. No arbitrary cap is needed, and any
// cap below it is a conformance break rather than a safety measure.
//
// Extract complete frames: [3-byte uint24 LE length][payload].
while (peer->recv_buf.size() >= FRAME_PREFIX_SIZE) {
uint32_t msg_len = read_frame_length(peer->recv_buf.data());
size_t total_frame = FRAME_PREFIX_SIZE + msg_len;
if (peer->recv_buf.size() < total_frame) break; // Incomplete frame.
// Extract the message payload.
transport->_dispatch_incoming(peer,
peer->recv_buf.data() + FRAME_PREFIX_SIZE, msg_len, PACKET_RELIABLE);
// Remove consumed frame from buffer.
peer->recv_buf.erase(peer->recv_buf.begin(),
peer->recv_buf.begin() + total_frame);
}
}
void UdxTransport::_on_stream_recv(udx_stream_t *stream, ssize_t read_len, const uv_buf_t *buf) {
if (read_len <= 0) return;
PeerConnection *peer = static_cast<PeerConnection *>(stream->data);
if (!peer) return;
UdxTransport *transport = static_cast<UdxTransport *>(stream->socket->data);
if (!transport) return;
// _on_stream_recv is always unreliable transport. For encrypted peers,
// the first byte is ciphertext — don't try to decode a header from it.
// _dispatch_incoming will decrypt and decode the header from plaintext.
transport->_dispatch_incoming(peer,
(const uint8_t *)buf->base, read_len, PACKET_UNRELIABLE);
}
void UdxTransport::_on_stream_close(udx_stream_t *stream, int status) {
PeerConnection *peer = static_cast<PeerConnection *>(stream->data);
if (!peer) return;
DHT_LOG("[UdxTransport] stream_close: peer=%d status=%d noise_ready=%d header_sent=%d header_recv=%d\n",
peer->godot_peer_id, status, peer->noise_ready, peer->header_sent, peer->header_received);
peer->connected = false;
// A retired peer (dedup, disconnect_peer) was already reported; its peer id
// may now belong to the connection that replaced it. Say nothing.
if (peer->doomed) return;
UdxTransport *transport = static_cast<UdxTransport *>(stream->socket->data);
if (!transport) {
DHT_LOG("[UdxTransport] stream_close: socket->data is NULL! Cannot propagate disconnect.\n");
return;
}
{
std::lock_guard<std::mutex> lock(transport->_events_mutex);
transport->_peer_events.push_back({peer->godot_peer_id, false});
if (transport->on_mux_gone) transport->on_mux_gone(peer->godot_peer_id);
}
// Outside the events lock: a consumer may react by writing or closing.
if (peer->stream_consumer && transport->_stream_close) transport->_stream_close(peer->godot_peer_id);
// Notify SwarmManager of disconnect for retry scheduling.
if (transport->_swarm_manager) {
transport->_swarm_manager->on_peer_disconnected(peer->godot_peer_id);
}
}
void UdxTransport::_on_stream_finalize(udx_stream_t *stream) {
// libudx has closed every handle in the stream: only now may its
// PeerConnection go, if it was retired.
PeerConnection *peer = static_cast<PeerConnection *>(stream->data);
if (!peer) return;
peer->finalized = true;
if (peer->doomed) delete peer;
}
void UdxTransport::_on_socket_recv(udx_socket_t *socket, ssize_t read_len,
const uv_buf_t *buf, const struct sockaddr *from) {
if (read_len <= 0) return;
UdxTransport *transport = static_cast<UdxTransport *>(socket->data);
if (!transport) return;
const uint8_t *data = (const uint8_t *)buf->base;
size_t len = (size_t)read_len;
const struct sockaddr_in *from_in = (const struct sockaddr_in *)from;
uint8_t from_ip[4];
memcpy(from_ip, &from_in->sin_addr.s_addr, 4);
int from_port = ntohs(from_in->sin_port);
// Dispatch DHT-RPC messages (type byte 0x03 = request, 0x13 = response).
if (transport->_dht_rpc && len >= 4) {
uint8_t type_byte = data[0];
if (type_byte == DHT_REQUEST_ID || type_byte == DHT_RESPONSE_ID) {
transport->_dht_rpc->on_message(data, len, from_in);
return;
}
}
// Dispatch holepunch probes (1-byte payload 0x00).
if (transport->_dht_connector && len == 1 && data[0] == 0x00) {
IPv4Address addr;
memcpy(addr.ip, from_ip, 4);
addr.port = from_port;
transport->_dht_connector->on_probe_received(addr);
return;
}
}
// udx: the write queue fell back under the high watermark.
void UdxTransport::_on_stream_drain(udx_stream_t *stream) {
PeerConnection *peer = (PeerConnection *)stream->data;
if (!peer) return;
peer->writable = true;
// Tell whoever parked. Protomux forwards it to its consumer — for the overlay
// that is the serve pump, which resumes from where it stopped.
if (peer->mux) peer->mux->notify_drain();
if (peer->stream_consumer) {
UdxTransport *transport = static_cast<UdxTransport *>(stream->socket->data);
if (transport && transport->_stream_drain) transport->_stream_drain(peer->godot_peer_id);
}
}
void UdxTransport::_on_write_ack(udx_stream_write_t *req, int status, int unordered) {
(void)status;
(void)unordered;
// Free the heap-allocated data copy attached in _send_raw / CMD_SEND.
if (req->data) free(req->data);
free(req);
}
void UdxTransport::_on_send_complete(udx_stream_send_t *req, int status) {
(void)status;
// Free the heap-allocated data copy attached in CMD_SEND (unreliable path).
if (req->data) free(req->data);
free(req);
}
// ---- Peer Management (UDX thread only, except where locked) ----
PeerConnection *UdxTransport::_get_peer(int32_t peer_id) {
auto it = _peers.find(peer_id);
return (it != _peers.end()) ? it->second : nullptr;
}
PeerConnection *UdxTransport::_add_peer(int32_t peer_id, uint32_t local_sid,
uint32_t remote_sid, const struct sockaddr *addr,
udx_socket_t *socket_override) {
PeerConnection *peer = new PeerConnection();
peer->godot_peer_id = peer_id;
peer->local_stream_id = local_sid;
peer->remote_stream_id = remote_sid;
memcpy(&peer->remote_addr, addr, sizeof(struct sockaddr_in));
// Initialize the UDX stream.
udx_stream_init(&_udx, &peer->stream, local_sid,
_on_stream_close, _on_stream_finalize);
peer->stream.data = peer;
peer->stream_initialized = true;
// Connect stream to remote — use birthday winner socket if available.
udx_socket_t *use_socket = socket_override ? socket_override : &_socket;
udx_stream_connect(&peer->stream, use_socket, remote_sid, addr);
// Keep an idle stream's NAT mapping open. Nothing called this before, and
// DhtConnector's comment at the point it frees a CONNECTED connection —
// "UDX stream takes over with its own keepalive" — described something that
// was therefore not happening.
//
// ⛔ IT IS NOT LIVENESS, AND IT CANNOT BE. libudx's send_probe writes a bare
// UDX_HEADER_HEARTBEAT with uv_udp_try_send; it never enters the packet or
// RTO machinery, and libudx says so itself at that line:
//
// // todo: send a data packet with seq=remote_acked-1 instead
//
// So an unanswered heartbeat produces no retransmit, never reaches
// UDX_MAX_RTO_TIMEOUTS, and never calls close_stream(UV_ETIMEDOUT). Only a
// stream with real data outstanding dies that way. An IDLE stream to a
// process that has been killed stays "connected" for ever — measured, in
// swarmd's gate D5: kill one of two meshed daemons and the survivor still
// reports peers=1 ninety seconds later.
//
// Detecting that is OURS to build, above this layer, and it is not built.
// See docs/SWARMD-DESIGN.md §8 (D5) and PARITY-hyperswarm.md §17.
udx_stream_set_keepalive(&peer->stream, (uint32_t)DHT_KEEPALIVE_INTERVAL_MS);
// Enable keepalive probes every 10s to detect dead connections.
udx_stream_set_keepalive(&peer->stream, 10000);
// Start reading — both reliable and unreliable.
udx_stream_read_start(&peer->stream, _on_stream_read);
udx_stream_recv_start(&peer->stream, _on_stream_recv);
_peers[peer_id] = peer;
return peer;
}
void UdxTransport::_remove_peer(int32_t peer_id) {
auto it = _peers.find(peer_id);
if (it != _peers.end()) {
// Remove from pubkey dedup map.
// Note: Protomux cleanup happens in ~PeerConnection (shutdown + delete).
PeerConnection *peer = it->second;
if (peer->encrypted) {
char hex_buf[65];
for (int i = 0; i < 32; i++) {
snprintf(hex_buf + i * 2, 3, "%02x", peer->remote_public_key[i]);
}
hex_buf[64] = '\0';
auto pk_it = _peers_by_pubkey.find(std::string(hex_buf));
if (pk_it != _peers_by_pubkey.end() && pk_it->second == peer_id) {
_peers_by_pubkey.erase(pk_it);
}
}
_peers.erase(it);
// Delete now only if libudx is done with the stream; otherwise retire it
// and let _on_stream_finalize delete it (see PeerConnection::doomed).
if (!peer->stream_initialized || peer->finalized) {
delete peer;
return;
}
peer->doomed = true;
if (peer->connected) {
udx_stream_destroy(&peer->stream);
peer->connected = false;
}
}
}
int32_t UdxTransport::_next_peer_id() {
return ++_peer_id_counter;
}
// ---- Protomux helpers (UDX thread) ----
// ⛔ THE LIVENESS THAT NOTHING ELSE PROVIDES. hyper-games #75 PART 3 traced the
// hole through three layers; this is the fix, and it is deliberately at OUR
// layer rather than libudx's, because libudx's heartbeat cannot be made to
// detect death without changing libudx.
//
// TWO HALVES, and the second is conditional on purpose:
//
// send an idle peer gets a ZERO-LENGTH encrypted frame every
// PEER_KEEPALIVE_INTERVAL_MS. That is the reference's wire form
// exactly — @hyperswarm/secret-stream's sendKeepAlive writes
// `this.alloc(0)` — so a JS peer eats ours and we eat theirs. A write
// of any kind counts, so a busy peer never sends one.
//
// kill a peer that has sent us nothing for PEER_DEAD_MS is disconnected,
// which drives the close path that already exists: _on_stream_close ->
// on_mux_gone -> PEER_DOWN.
//
// ⛔ THE KILL ONLY ARMS ONCE THE PEER HAS SENT A KEEPALIVE ITSELF. A build that
// predates this sends nothing while idle, and killing it after twenty seconds
// would make an idle connection a flag day — the one thing the swarmd-session
// work went out of its way to avoid. Until a peer proves it speaks keepalives,
// its silence means what it meant before: nothing.
void UdxTransport::_sweep_keepalive(uint64_t now_ms) {
_loop_now_ms = now_ms;
std::vector<int32_t> dead;
for (auto &kv : _peers) {
PeerConnection *peer = kv.second;
if (!peer || !peer->connected || !peer->noise_ready) continue;
// A peer that has never been heard from starts its clock now rather than
// at zero, or it is declared dead the moment it connects.
if (peer->last_recv_ms == 0) peer->last_recv_ms = now_ms;
if (peer->last_send_ms == 0) peer->last_send_ms = now_ms;
if (peer->peer_keepalive && now_ms - peer->last_recv_ms > PEER_DEAD_MS) {
// ⛔ NOT DHT_LOG. Dropping a peer for silence is a decision this
// layer makes ON ITS OWN, and it is indistinguishable from the peer
// leaving unless it says so. Compiled out, it is exactly the kind of
// invisible verdict that made "joined" mean nothing.
fprintf(stderr, "[UdxTransport] peer %d silent for %llums — dropping it\n",
peer->godot_peer_id,
(unsigned long long)(now_ms - peer->last_recv_ms));
dead.push_back(kv.first);
continue;
}
if (now_ms - peer->last_send_ms >= PEER_KEEPALIVE_INTERVAL_MS) {
_send_encrypted(peer, nullptr, 0); // stamps last_send_ms
peer->keepalives_sent++;
// Once, when the first one goes out: enough to prove the mechanism
// is running without a line every five seconds per peer.
if (peer->keepalives_sent == 1) {
fprintf(stderr, "[UdxTransport] peer %d: keepalives started\n",
peer->godot_peer_id);
}
}
}
// Queued rather than done here: CMD_DISCONNECT already destroys the stream,
// pushes the peer event and removes the peer — a tested path — and doing it
// inline would mutate _peers while this loop walks it. It runs on the next
// tick, one millisecond later.
for (int32_t id : dead) disconnect_peer(id, true);
}
void UdxTransport::_send_encrypted(PeerConnection *peer, const uint8_t *data, size_t len) {
DHT_LOG("[UdxTransport] _send_encrypted: %zu bytes for peer %d (noise=%p noise_ready=%d)\n",
len, peer->godot_peer_id, (void *)peer->noise, peer->noise_ready);
if (!peer->noise || !peer->noise_ready) return;
std::vector<uint8_t> encrypted = peer->noise->encrypt(data, len);
DHT_LOG("[UdxTransport] _send_encrypted: encrypted to %zu bytes\n", encrypted.size());
size_t framed_len = FRAME_PREFIX_SIZE + encrypted.size();
uint8_t *copy = (uint8_t *)malloc(framed_len);
write_frame_length(copy, static_cast<uint32_t>(encrypted.size()));
memcpy(copy + FRAME_PREFIX_SIZE, encrypted.data(), encrypted.size());
uv_buf_t buf = uv_buf_init((char *)copy, static_cast<unsigned int>(framed_len));
udx_stream_write_t *req = (udx_stream_write_t *)malloc(udx_stream_write_sizeof(1));
req->data = copy;
// KEEP THE RETURN. udx_stream_write answers 1 while the stream can take more
// and 0 once writes_queued_bytes passes the high watermark; negative is an
// error. Discarding it left every protocol above unable to tell a healthy
// stream from one it was burying, so a bulk sender (the overlay's chunk pump)
// had nothing to throttle against.
const int rc = udx_stream_write(req, &peer->stream, &buf, 1, _on_write_ack);
peer->last_send_ms = _loop_now_ms; // a write IS a keepalive; see _sweep_keepalive
peer->writable = rc > 0;
if (rc == 0) {
// Arm the drain edge exactly once per peer. udx calls on_drain when the
// queue falls back under the watermark, which is the signal a parked
// producer resumes on.
if (!peer->drain_armed) {
peer->drain_armed = true;
peer->stream.data = peer;
udx_stream_write_resume(&peer->stream, _on_stream_drain);
}
}
}
void UdxTransport::_deliver_godot_packet(PeerConnection *peer,
const uint8_t *data, size_t len) {
Packet pkt(data, static_cast<int32_t>(len), peer->godot_peer_id, 0, PACKET_RELIABLE);
_incoming.push(std::move(pkt));
}
// ---- Noise Handshake / Header Handling (UDX thread) ----
void UdxTransport::_send_raw(PeerConnection *peer, const uint8_t *data, size_t len) {
// Length-prefixed framing: [3-byte uint24 LE length][payload].
// UDX reliable streams are byte streams — writes can coalesce into one read.
size_t framed_len = FRAME_PREFIX_SIZE + len;
uint8_t *copy = (uint8_t *)malloc(framed_len);
write_frame_length(copy, static_cast<uint32_t>(len));
memcpy(copy + FRAME_PREFIX_SIZE, data, len);
uv_buf_t buf = uv_buf_init((char *)copy, static_cast<unsigned int>(framed_len));
udx_stream_write_t *req = (udx_stream_write_t *)malloc(udx_stream_write_sizeof(1));
req->data = copy; // Attach so _on_write_ack can free it.
udx_stream_write(req, &peer->stream, &buf, 1, _on_write_ack);
}
void UdxTransport::_handle_noise_message(PeerConnection *peer, NoiseMessageType msg_type,
const uint8_t *payload, size_t payload_len) {
if (msg_type == NOISE_MSG_HANDSHAKE) {
if (!peer->handshake) {
// Server received first handshake message — create responder state.
peer->handshake = new NoiseHandshake(false, _local_keypair);
peer->noise = new NoiseStream();
peer->is_noise_initiator = false;
}
// Process the received handshake message.
auto result = peer->handshake->recv(payload, payload_len);
// If handshake is not complete, send our response.
if (!peer->handshake->is_complete()) {
auto reply = peer->handshake->send();
std::vector<uint8_t> framed(1 + reply.size());
framed[0] = NOISE_MSG_HANDSHAKE;
memcpy(framed.data() + 1, reply.data(), reply.size());
_send_raw(peer, framed.data(), framed.size());
}
if (peer->handshake->is_complete()) {
// Save remote public key.
memcpy(peer->remote_public_key, peer->handshake->remote_public_key(), 32);
// Store rx_key before handshake deletion.
memcpy(peer->_rx_key, peer->handshake->rx_key(), 32);
// Save handshake hash and compute streamId.
memcpy(peer->_handshake_hash, peer->handshake->handshake_hash(), 64);
compute_stream_id(peer->_stream_id, peer->_handshake_hash, peer->is_noise_initiator);
// Initialise TX stream encryption.
peer->noise->init_tx(peer->handshake->tx_key());
// Send our TX header: [32B streamId][24B XChaCha20 header] = 56 bytes.
const uint8_t *hdr = peer->noise->tx_header();
std::vector<uint8_t> hdr_msg(32 + NoiseStream::HEADER_BYTES);
memcpy(hdr_msg.data(), peer->_stream_id, 32);
memcpy(hdr_msg.data() + 32, hdr, NoiseStream::HEADER_BYTES);
_send_raw(peer, hdr_msg.data(), hdr_msg.size());
peer->header_sent = true;
// Clean up handshake.
delete peer->handshake;
peer->handshake = nullptr;
}
} else if (msg_type == NOISE_MSG_HEADER) {
// Received remote's TX header — initialise our RX state.
// Format: [32B streamId][24B XChaCha20 header] = 56 bytes.
// Guard: only process once (matches JS secret-stream _setup flag).
if (peer->header_received) return; // Already received — ignore duplicate.
if (payload_len < (size_t)(32 + NoiseStream::HEADER_BYTES)) {
DHT_LOG("[UdxTransport] noise_header: REJECT — too short (%zu < %d)\n",
payload_len, 32 + NoiseStream::HEADER_BYTES);
return;
}
DHT_LOG("[UdxTransport] noise_header: received RX header from peer %d (%zu bytes, streamId=%02x%02x...)\n",
peer->godot_peer_id, payload_len, payload[0], payload[1]);
// Skip 32-byte streamId, extract 24-byte header.
const uint8_t *header = payload + 32;
// Initialise RX with stored key + remote's TX header.
peer->noise->init_rx(peer->_rx_key, header);
peer->header_received = true;
if (peer->header_sent && peer->header_received) {
peer->noise_ready = true;
DHT_LOG("[UdxTransport] noise_ready! peer=%d — encryption fully established\n",
peer->godot_peer_id);
// A stream consumer takes the connection whole: no internal mux, no
// godot-rpc. See set_stream_consumer.
if (_stream_open) {
peer->stream_consumer = true;
{
std::lock_guard<std::mutex> lock(_events_mutex);
_peer_events.push_back({peer->godot_peer_id, true});
}
_stream_open(stream_info_of(peer));
return;
}
// Create Protomux for this peer.
DHT_LOG("[UdxTransport] creating Protomux for peer %d\n", peer->godot_peer_id);
peer->mux = new Protomux();
peer->mux->set_write_callback([this, peer](const uint8_t *data, size_t len) {
_send_encrypted(peer, data, len);
return peer->writable;
});
// Auto-create godot-rpc channel (channel 0).
DHT_LOG("[UdxTransport] creating godot-rpc channel for peer %d\n", peer->godot_peer_id);
auto *godot_ch = peer->mux->create_channel("godot-rpc");
godot_ch->add_message([this, peer](const uint8_t *data, size_t len, ProtomuxChannel *) {
_deliver_godot_packet(peer, data, len);
});
godot_ch->on_open = [this, peer](const uint8_t *, size_t) {
// Channel fully open — NOW emit peer_connected to Godot.
DHT_LOG("[UdxTransport] protomux godot-rpc channel open for peer %d\n",
peer->godot_peer_id);
{
std::lock_guard<std::mutex> lock(_events_mutex);
_peer_events.push_back({peer->godot_peer_id, true});
}
// Tell any consumer protocol the mux is usable. On the UDX thread,
// which is where it must attach its channel. Recorded as well as
// fired, so a consumer that registers later is told the same thing
// rather than missing this peer entirely — see set_mux_callbacks.
peer->mux_ready = true;
if (on_mux_ready) on_mux_ready(peer->godot_peer_id, peer->mux);
};
godot_ch->open();
}
}
}
void UdxTransport::_queue_command(Command &&cmd) {
std::lock_guard<std::mutex> lock(_cmd_mutex);
_commands.push_back(std::move(cmd));
}
// ---- Swarm Manager API (called from Godot main thread) ----
void UdxTransport::swarm_join(const uint8_t *topic, bool server, bool client) {
Command cmd;
cmd.type = CMD_SWARM_JOIN;
cmd.data.assign(topic, topic + 32);
cmd.server = server;
cmd.client = client;
_queue_command(std::move(cmd));
_wake();
}
void UdxTransport::swarm_leave(const uint8_t *topic) {
Command cmd;
cmd.type = CMD_SWARM_LEAVE;
cmd.data.assign(topic, topic + 32);
_queue_command(std::move(cmd));
_wake();
}
void UdxTransport::swarm_join_peer(const uint8_t *pk) {
Command cmd;
cmd.type = CMD_SWARM_JOIN_PEER;
memcpy(cmd.remote_public_key, pk, 32);
_queue_command(std::move(cmd));
_wake();
}
void UdxTransport::swarm_leave_peer(const uint8_t *pk) {
Command cmd;
cmd.type = CMD_SWARM_LEAVE_PEER;
memcpy(cmd.remote_public_key, pk, 32);
_queue_command(std::move(cmd));
_wake();
}
bool UdxTransport::pop_swarm_event(SwarmEvent &out) {
std::lock_guard<std::mutex> lock(_events_mutex);
if (_swarm_events.empty()) return false;
out = _swarm_events.front();
_swarm_events.pop_front();
return true;
}
SwarmStatus UdxTransport::get_swarm_status() const {
if (_swarm_manager) {
return _swarm_manager->get_status();
}
// Return zeroed status when SwarmManager not yet initialized.
SwarmStatus s;
s.peers_known = 0;
s.peers_connected = 0;
s.connecting = 0;
s.queue_size = 0;
s.max_peers = SWARM_MAX_PEERS;
s.max_parallel = SWARM_MAX_PARALLEL;
s.explicit_count = 0;
return s;
}
bool UdxTransport::get_peer_public_key(int32_t peer_id, uint8_t *out_pk) const {
auto it = _peers.find(peer_id);
if (it == _peers.end() || !it->second) return false;
memcpy(out_pk, it->second->remote_public_key, 32);
return true;
}
bool UdxTransport::get_peer_rtt(int32_t peer_id, PeerRttStats &out) const {
auto it = _peers.find(peer_id);
if (it == _peers.end() || !it->second) return false;
const udx_stream_t &s = it->second->stream;
out.srtt_ms = s.srtt;
out.rttvar_ms = s.rttvar;
out.rto_ms = s.rto;
out.min_rtt_ms = s.bbr.min_rtt_ms;
return true;
}
// ---- DHT Integration (UDX thread only) ----
void UdxTransport::_init_dht() {
if (_dht_rpc) return; // Already initialized.
// Create DHT-RPC client.
_dht_rpc = new DhtRpc();
_dht_rpc->init(&_socket, &_loop);
// Built HERE and thrown away: set_bootstrap parses each string into an
// IPv4Address on the spot, so the pointers only have to survive the call.
// Building the array in the same pass that fills the strings is what does not
// work — a vector of short strings keeps them inside its own element block, so
// each growth frees the block that every c_str() taken so far pointed into.
std::vector<const char *> ptrs;
ptrs.reserve(_bootstrap.size());
for (const std::string &s : _bootstrap) ptrs.push_back(s.c_str());
_dht_rpc->set_bootstrap(ptrs.data(), (int)ptrs.size());
// Get bound port from getsockname (needed for DhtRpc local address).
struct sockaddr_in local_addr;
int addr_len = sizeof(local_addr);
udx_socket_getsockname(&_socket, (struct sockaddr *)&local_addr, &addr_len);
uint16_t bound_port = ntohs(local_addr.sin_port);
// Set DhtRpc local address (bind addr + port).
IPv4Address rpc_local;
memcpy(rpc_local.ip, &local_addr.sin_addr.s_addr, 4);
rpc_local.port = bound_port;
_dht_rpc->set_local_address(rpc_local);
// Collect real interface addresses via uv_interface_addresses().
// These are passed to both DhtRpc (for handshake payload) and DhtConnector.
// This replaces the old getsockname() approach which only returned 0.0.0.0.
uv_interface_address_t *ifaces = nullptr;
int iface_count = 0;
uv_interface_addresses(&ifaces, &iface_count);
IPv4Address lan_addrs[HP_MAX_ADDRESSES];
int lan_count = 0;
if (ifaces) {
for (int i = 0; i < iface_count && lan_count < HP_MAX_ADDRESSES; i++) {
if (ifaces[i].is_internal) continue; // skip loopback
if (ifaces[i].address.address4.sin_family != AF_INET) continue; // IPv4 only
memcpy(lan_addrs[lan_count].ip, &ifaces[i].address.address4.sin_addr.s_addr, 4);
lan_addrs[lan_count].port = bound_port;
lan_count++;
}
uv_free_interface_addresses(ifaces, iface_count);
}
// Fallback: if no interfaces found, use the bind address.
if (lan_count == 0) {
lan_addrs[0] = rpc_local;
lan_count = 1;
}
DHT_LOG("[UdxTransport] _init_dht: %d LAN addresses (port %d)\n", lan_count, bound_port);
for (int i = 0; i < lan_count; i++) {
DHT_LOG("[UdxTransport] iface[%d] = %d.%d.%d.%d:%d\n", i,
lan_addrs[i].ip[0], lan_addrs[i].ip[1],
lan_addrs[i].ip[2], lan_addrs[i].ip[3], lan_addrs[i].port);
}
// Pass LAN IPs to DhtRpc so handshake NoisePayload.addresses4 includes them.
// This matches JS behavior where Holepuncher.localAddresses() adds LAN IPs.
_dht_rpc->set_interface_addresses(lan_addrs, lan_count);
// Create HyperDHT layer.
_hyperdht = new HyperDht();
_hyperdht->init(_dht_rpc, _dht_keypair);
// Create connection manager.
_dht_connector = new DhtConnector();
_dht_connector->init(_hyperdht, &_socket, &_loop);
// Set local addresses for holepunch advertisements.
_dht_connector->set_local_addresses(lan_addrs, lan_count);
// Register callbacks.
_dht_connector->on_connected = [this](uint32_t conn_id, const DhtConnection &conn) {
_dht_on_connected(conn_id, conn);
};
_dht_connector->on_failed = [this](uint32_t conn_id, const uint8_t *remote_pk, const char *error) {
if (_swarm_manager) {
_swarm_manager->on_connect_failed(remote_pk);
}
};
// ⛔ TELL THE CONNECTOR WHEN A QUERY EXHAUSTS. Without this it learns that a
// peer is unreachable only from DHT_CONNECT_TIMEOUT_MS, 30 seconds after the
// DHT already said so, and holds one of SWARM_MAX_PARALLEL slots the whole
// time. JS has no overall connect timer at all — its find_peer stage simply
// ends when the query does. See docs/PARITY-hyperswarm.md §11.2.
_hyperdht->on_query_complete = [this](uint32_t query_id) {
if (_dht_connector) _dht_connector->on_find_peer_complete(query_id);
};
// Hook holepunch callbacks — route DHT events to the connector.
_hyperdht->on_holepunch_request = [this](const DhtRequest &req) {
if (_dht_connector) {
_dht_connector->on_holepunch_message(req);
}
};
_hyperdht->on_holepunch_response = [this](uint32_t tag, const uint8_t *value,
size_t value_len, const IPv4Address &from) {
if (_dht_connector) {
_dht_connector->on_holepunch_response(tag, value, value_len, from);
}
};
// Create SwarmManager (after connector is ready).
_swarm_manager = new SwarmManager();
_swarm_manager->init(_dht_connector, _hyperdht, _dht_keypair.public_key);
// Hook peer found events for signal emission to Godot.
// DhtConnector::init already sets _hyperdht->on_peer_found,
// so we wrap it to also push events to the main thread
// AND route to SwarmManager for auto-connect.
auto existing_peer_found = _hyperdht->on_peer_found;
_hyperdht->on_peer_found = [this, existing_peer_found](uint32_t query_id, const FoundPeer &peer) {
// Push event for Godot signal (backward compat: dht_peer_found).
{
std::lock_guard<std::mutex> lock(_events_mutex);
PeerFoundEvent evt;
memcpy(evt.public_key, peer.public_key, 32);
_peer_found_events.push_back(evt);
}
// Route to SwarmManager for auto-connect, with topic for retry.
if (_swarm_manager) {
const uint8_t *topic = _hyperdht->get_query_topic(query_id);
_swarm_manager->on_peer_found(peer.public_key, topic);
}
// Forward to connector's handler.
if (existing_peer_found) existing_peer_found(query_id, peer);
};
// Hook DHT storage callbacks for Godot signal emission.
_hyperdht->on_immutable_put_complete = [this](const uint8_t *key) {
std::lock_guard<std::mutex> lock(_events_mutex);
DhtStorageEvent evt;
evt.type = DhtStorageEvent::IMMUTABLE_PUT_COMPLETE;
memcpy(evt.key, key, 32);
evt.seq = 0;
evt.success = true;
_storage_events.push_back(std::move(evt));
};
_hyperdht->on_immutable_get_complete = [this](const uint8_t *key,
const uint8_t *value, size_t value_len) {
std::lock_guard<std::mutex> lock(_events_mutex);
DhtStorageEvent evt;
evt.type = DhtStorageEvent::IMMUTABLE_GET_COMPLETE;
memcpy(evt.key, key, 32);
if (value && value_len > 0) {
evt.value.assign(value, value + value_len);
evt.success = true;
} else {
evt.success = false;
}
evt.seq = 0;
_storage_events.push_back(std::move(evt));
};
_hyperdht->on_mutable_put_complete = [this](bool success) {
std::lock_guard<std::mutex> lock(_events_mutex);
DhtStorageEvent evt;
evt.type = DhtStorageEvent::MUTABLE_PUT_COMPLETE;
memset(evt.key, 0, 32);
evt.seq = 0;
evt.success = success;
_storage_events.push_back(std::move(evt));
};
_hyperdht->on_mutable_get_complete = [this](const uint8_t *public_key,
const uint8_t *value, size_t value_len,
uint64_t seq) {
std::lock_guard<std::mutex> lock(_events_mutex);
DhtStorageEvent evt;
evt.type = DhtStorageEvent::MUTABLE_GET_COMPLETE;
memcpy(evt.key, public_key, 32);
if (value && value_len > 0) {
evt.value.assign(value, value + value_len);
evt.success = true;
} else {
evt.success = false;
}
evt.seq = seq;
_storage_events.push_back(std::move(evt));
};
// Bootstrap into the DHT network.
_dht_rpc->on_bootstrapped = [this]() {
_dht_bootstrapped.store(true);
};
_dht_rpc->bootstrap();
_dht_active.store(true);
}
void UdxTransport::_dht_on_connected(uint32_t conn_id, const DhtConnection &conn) {
// Create a sockaddr from the punched/connected address.
struct sockaddr_in addr;
memset(&addr, 0, sizeof(addr));
addr.sin_family = AF_INET;
memcpy(&addr.sin_addr.s_addr, conn.punched_address.ip, 4);
addr.sin_port = htons(conn.punched_address.port);
// local_sid IS the handshake id — that is what we advertised, so print the one
// that is actually on the wire. This line used to print a separate
// DhtConnection::local_stream_id that nothing sent and nothing listened on; it
// read as a stream-id mismatch on every healthy connection and sent the #15
// diagnosis chasing the wrong layer. It is a pairing when hs_id and remote_sid
// belong to the same exchange, and `initiator` is what says whether they should.
DHT_LOG("[UdxTransport] _dht_on_connected: conn=%u addr=%d.%d.%d.%d:%d "
"hs_id/local_sid=%u remote_sid=%u initiator=%d punch_socket=%s\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,
conn.handshake_id, conn.remote_stream_id, (int)conn.is_initiator,
conn.punch_socket ? "birthday" : "main");
// ---- Birthday socket setup ----
// If the holepunch was won by a birthday socket, set its data pointer
// to this transport so UDX stream callbacks can access us.
// Without this, _on_stream_read silently drops all data (socket->data is null).
if (conn.punch_socket) {
conn.punch_socket->data = this;
}
// ---- Connection dedup (mesh support) ----
// Convert remote public key to hex string for map lookup.
char hex_buf[65];
for (int i = 0; i < 32; i++) {
snprintf(hex_buf + i * 2, 3, "%02x", conn.remote_public_key[i]);
}
hex_buf[64] = '\0';
std::string remote_pk_hex(hex_buf);
auto existing = _peers_by_pubkey.find(remote_pk_hex);
if (existing != _peers_by_pubkey.end()) {
int32_t old_peer_id = existing->second;
PeerConnection *old_peer = _get_peer(old_peer_id);
// If the existing connection is fully established (noise_ready), never
// destroy it. Matches JS: existingIsOutdated check.
if (old_peer && old_peer->connected && old_peer->noise_ready) {
DHT_LOG("[UdxTransport] _dht_on_connected: DEDUP — keeping existing noise_ready peer %d\n", old_peer_id);
return; // Keep working connection, discard incoming.
}
// Both connections are still in setup. Deterministic tie-breaker:
// JS: expectedInitiator = b4a.compare(publicKey, remotePublicKey) > 0
// keepNew = existingIsOutdated || expectedInitiator === conn.isInitiator
bool we_are_greater = memcmp(_dht_keypair.public_key, conn.remote_public_key, 32) > 0;
bool keep_new = (we_are_greater == conn.is_initiator);
if (keep_new) {
// Destroy old (not-yet-ready) connection, proceed with new one.
DHT_LOG("[UdxTransport] _dht_on_connected: DEDUP — replacing old peer %d\n", old_peer_id);
if (old_peer && old_peer->connected) {
udx_stream_destroy(&old_peer->stream);
old_peer->connected = false;
{
std::lock_guard<std::mutex> lock(_events_mutex);
_peer_events.push_back({old_peer_id, false});
}
}
_remove_peer(old_peer_id);
} else {
// Keep old connection, discard new one.
DHT_LOG("[UdxTransport] _dht_on_connected: DEDUP — keeping old peer %d (discarding new)\n", old_peer_id);
return;
}
}
// Derive Godot peer ID from remote public key — globally consistent.
int32_t peer_id = pk_to_peer_id(conn.remote_public_key);
// Use the handshake_id as our local stream ID. This is what we sent
// in our NoisePayload (udx.id = hs->id), so the remote side expects it.
uint32_t local_sid = conn.handshake_id;
DHT_LOG("[UdxTransport] creating UDX stream: peer=%d local_sid=%u remote_sid=%u\n",
peer_id, local_sid, conn.remote_stream_id);
PeerConnection *peer = _add_peer(peer_id, local_sid, conn.remote_stream_id,
(struct sockaddr *)&addr, conn.punch_socket);
if (!peer) return;
peer->connected = true;
peer->encrypted = true;
peer->is_noise_initiator = conn.is_initiator;
memcpy(peer->remote_public_key, conn.remote_public_key, 32);
// Track by public key for dedup.
_peers_by_pubkey[remote_pk_hex] = peer_id;
// Copy handshake hash and compute streamId for header message.
memcpy(peer->_handshake_hash, conn.handshake_hash, 64);
compute_stream_id(peer->_stream_id, peer->_handshake_hash, peer->is_noise_initiator);
// Initialize NoiseStream encryption using DHT-derived keys.
peer->noise = new NoiseStream();
peer->noise->init_tx(conn.tx_key);
// Send our TX header: [32B streamId][24B XChaCha20 header] = 56 bytes.
const uint8_t *hdr = peer->noise->tx_header();
std::vector<uint8_t> hdr_msg(32 + NoiseStream::HEADER_BYTES);
memcpy(hdr_msg.data(), peer->_stream_id, 32);
memcpy(hdr_msg.data() + 32, hdr, NoiseStream::HEADER_BYTES);
_send_raw(peer, hdr_msg.data(), hdr_msg.size());
peer->header_sent = true;
DHT_LOG("[UdxTransport] sent TX header (%zu bytes, streamId=%02x%02x...) to peer %d\n",
hdr_msg.size(), peer->_stream_id[0], peer->_stream_id[1], peer_id);
// Store RX key — will be used when we receive remote's TX header.
memcpy(peer->_rx_key, conn.rx_key, 32);
// Note: peer_connected event fires when noise_ready becomes true
// (after both TX and RX headers are exchanged in _handle_noise_message).
// Notify SwarmManager of the new connection.
if (_swarm_manager) {
_swarm_manager->on_peer_connected(peer_id, conn.remote_public_key, conn.is_initiator);
}
}