#ifndef HYPERSWARM_UDX_TRANSPORT_H
#define HYPERSWARM_UDX_TRANSPORT_H
#include "dht_connector.h"
#include "dht_rpc.h"
#include "hyperdht.h"
#include "noise_handshake.h"
#include "noise_ik.h"
#include "noise_stream.h"
#include "packet_queue.h"
#include "protomux.h"
#include "swarm_manager.h"
#include <atomic>
#include <cstdint>
#include <deque>
#include <functional>
#include <map>
#include <mutex>
#include <string>
#include <thread>
#include <vector>
extern "C" {
#include <udx.h>
#include <uv.h>
}
// Internal message types for Noise handshake framing.
// First byte of every UDX payload distinguishes handshake from data.
enum NoiseMessageType : uint8_t {
NOISE_MSG_HANDSHAKE = 0xFE, // Handshake message
NOISE_MSG_HEADER = 0xFD, // Stream header exchange (post-handshake)
NOISE_MSG_DATA = 0x00, // Normal data (default, or any value != FE/FD)
};
// Derive a deterministic Godot peer ID from an Ed25519 public key.
// Both sides compute the same ID for the same key, so MultiplayerSynchronizer
// authority matches across machines.
// Returns a value in [2, INT32_MAX] — avoids 0 (broadcast) and 1 (server).
static inline int32_t pk_to_peer_id(const uint8_t *pk) {
uint32_t raw = ((uint32_t)pk[0] << 24) | ((uint32_t)pk[1] << 16) |
((uint32_t)pk[2] << 8) | (uint32_t)pk[3];
return 2 + (int32_t)(raw % (INT32_MAX - 1));
}
// Per-peer connection state managed by UdxTransport.
// ⛔ LIVENESS, because nothing else provides it. hyper-games #74/#75 traced this
// through three layers: DhtConnector's keepalive branch is unreachable (the
// connection is freed the instant it reaches CONNECTED), and libudx's keepalive
// sends a bare heartbeat OUTSIDE the packet and RTO machinery — its own TODO
// says so — so an unanswered one never reaches close_stream(UV_ETIMEDOUT). An
// idle stream to a killed process stayed "connected" for ever.
//
// The wire form is the reference's: @hyperswarm/secret-stream's `sendKeepAlive`
// writes a ZERO-LENGTH encrypted frame, and the receiver eats it
// (`if (plain.byteLength === 0 && this.keepAlive !== 0) return`).
static constexpr uint64_t PEER_KEEPALIVE_INTERVAL_MS = 5000; // hyperdht connectionKeepAlive
// Four consecutive keepalives missed. Long enough that a stalled loop or a
// hiccup does not cost a healthy peer, short enough to be useful.
static constexpr uint64_t PEER_DEAD_MS = 20000;
struct PeerConnection {
udx_stream_t stream;
uint32_t local_stream_id;
uint32_t remote_stream_id;
struct sockaddr_storage remote_addr;
bool connected;
int32_t godot_peer_id; // Godot-visible peer ID
// Noise encryption state.
NoiseHandshake *handshake; // Non-null during handshake phase.
NoiseStream *noise; // Non-null after handshake complete.
bool encrypted; // True if this is an encrypted connection.
bool noise_ready; // True when both TX and RX streams are initialised.
bool is_noise_initiator; // True if this peer initiated the Noise handshake.
bool header_sent; // True after we've sent our TX header.
bool header_received; // True after we've received remote's TX header.
uint8_t remote_public_key[32]; // Remote peer's verified static public key.
uint8_t _rx_key[32]; // Stored RX key (saved before handshake deletion).
uint8_t _handshake_hash[64]; // Noise handshake hash (for streamId derivation).
uint8_t _stream_id[32]; // SecretStream streamId (sent in header message).
// Write-side flow control. `writable` is false between udx reporting its high
// watermark and the drain callback firing; `drain_armed` keeps the resume
// registration to once per peer.
bool writable = true;
bool drain_armed = false;
// ---- liveness (see UdxTransport::_sweep_keepalive) ----
// When we last received an AUTHENTICATED frame from this peer, and when we
// last wrote to it. Both in the UDX loop's clock.
uint64_t last_recv_ms = 0;
uint64_t last_send_ms = 0;
// True once this peer has sent an empty keepalive frame. Until it does, its
// silence is not evidence of death — it may simply be a build that predates
// keepalives, and killing it would be a flag day for idle connections.
bool peer_keepalive = false;
uint64_t keepalives_sent = 0;
// Protomux multiplexer (created after Noise handshake completes).
Protomux *mux = nullptr;
// True once the godot-rpc channel has OPENED — the same instant on_mux_ready
// fires. A late consumer is replayed against this rather than against `mux`
// being non-null, so it is told exactly what a punctual one would have been.
bool mux_ready = false;
// True when this connection belongs to the stream consumer (no internal mux).
bool stream_consumer = false;
// ⛔ LIFETIME. udx_stream_t (held by value above) embeds libuv handles that
// udx_stream_destroy closes ASYNCHRONOUSLY, signalling completion via the
// stream's on_finalize. A PeerConnection whose stream is initialised must
// therefore not be deleted before finalize — _remove_peer RETIRES it (doomed)
// and _on_stream_finalize deletes it. Deleting at once left libuv walking its
// closing queue through freed memory: uv__finish_close's assertion.
bool stream_initialized = false; // udx_stream_init ran
bool finalized = false; // libudx finished with the stream's handles
bool doomed = false; // removed from _peers; delete on finalize
// Receive buffer for length-prefixed framing (matches JS SecretStream's uint24 LE).
// UDX reliable streams are byte streams — multiple writes can coalesce into one read.
std::vector<uint8_t> recv_buf;
PeerConnection()
: local_stream_id(0), remote_stream_id(0), connected(false), godot_peer_id(0),
handshake(nullptr), noise(nullptr), encrypted(false), noise_ready(false),
is_noise_initiator(false), header_sent(false), header_received(false) {
memset(&remote_addr, 0, sizeof(remote_addr));
memset(&stream, 0, sizeof(stream));
memset(remote_public_key, 0, 32);
memset(_rx_key, 0, 32);
memset(_handshake_hash, 0, 64);
memset(_stream_id, 0, 32);
}
~PeerConnection() {
delete handshake;
delete noise;
if (mux) {
mux->shutdown();
delete mux;
}
}
};
// Peer event passed from UDX thread to Godot main thread.
struct PeerEvent {
int32_t peer_id;
bool connected; // true = connected, false = disconnected
};
// Peer found event from DHT discovery (UDX thread → main thread).
struct PeerFoundEvent {
uint8_t public_key[32];
};
// DHT storage event from DHT operations (UDX thread → main thread).
struct DhtStorageEvent {
enum Type {
IMMUTABLE_PUT_COMPLETE,
IMMUTABLE_GET_COMPLETE,
MUTABLE_PUT_COMPLETE,
MUTABLE_GET_COMPLETE,
};
Type type;
uint8_t key[32]; // For immutable: BLAKE2b(value). For mutable: publicKey.
std::vector<uint8_t> value; // Retrieved value (empty if not found).
uint64_t seq; // Sequence number (mutable only).
bool success; // True if operation succeeded / value found.
};
// C++ wrapper around libudx.
// Runs the libuv event loop in a background thread.
// All public methods are thread-safe (called from Godot's main thread).
class UdxTransport {
public:
using PeerConnectedCallback = std::function<void(int32_t peer_id)>;
using PeerDisconnectedCallback = std::function<void(int32_t peer_id)>;
UdxTransport();
~UdxTransport();
// Start as server — bind to port, accept incoming streams (unencrypted).
bool start_server(int port);
// Start as client — connect to a server at addr:port (unencrypted).
bool start_client(const char *address, int port);
// Start as server with Noise encryption.
bool start_server_encrypted(int port, const NoiseKeypair &keypair);
// Start as client with Noise encryption + remote public key verification.
bool start_client_encrypted(const char *address, int port,
const uint8_t *remote_public_key);
// Stop everything and join the background thread.
void stop();
// Send a packet to a specific peer or broadcast (peer_id=0).
void send(int32_t target_peer, const uint8_t *data, int32_t size,
PacketTransferMode mode, int32_t channel);
// Disconnect a specific peer.
void disconnect_peer(int32_t peer_id, bool force);
// Read incoming packets (called from Godot main thread).
PacketQueue &incoming() { return _incoming; }
const PacketQueue &incoming() const { return _incoming; }
// Process commands queued from the main thread.
// Public because it's called from the libuv timer callback.
void _process_commands();
// UDX THREAD. Send a keepalive to any idle peer, and disconnect any peer that
// has stopped answering. Public for the same reason _process_commands is: the
// loop's timer callback is a free function.
void _sweep_keepalive(uint64_t now_ms);
public:
// ---- UDX-thread execution seam -----------------------------------------
//
// udx and protomux handles belong to this loop's thread. Anything that writes
// them must run here, which is why every cross-thread operation in this class
// is already queued rather than called. `run_on_loop` generalises that for
// consumers outside it — the overlay's fetch/serve commands arrive on Godot's
// main thread and must not touch a socket from there.
void run_on_loop(std::function<void()> task);
// Peer lifecycle, fired ON THE UDX THREAD once a peer's Protomux exists and
// again when it goes away. A consumer that wants to speak its own protocol
// over the connection registers here and gets the mux to attach a channel to.
//
// REGISTER THROUGH set_mux_callbacks, NOT by assigning these directly. Two
// reasons, and the first one cost a day:
//
// 1. Registration must be RETROACTIVE. A consumer that attaches after the
// swarm has meshed — the normal ordering, since you wait for peers before
// you want to talk to them — would otherwise learn about no peer that
// already exists, and stay silent until the swarm happened to churn. That
// wait is unbounded: layer 12 measured 19.4s on one run and never on the
// next, and the resulting flake read as "the overlay channel is slow to
// open" rather than "the peer was never announced".
// 2. Assigning a std::function from Godot's thread while this one may be
// calling it is a data race. The setter marshals the assignment onto the
// loop, where every other use of these already lives.
std::function<void(int32_t peer_id, Protomux *mux)> on_mux_ready;
std::function<void(int32_t peer_id)> on_mux_gone;
// Set (or clear, with nullptr) the pair above on the loop thread, then replay
// on_mux_ready for every peer whose mux is already usable.
void set_mux_callbacks(std::function<void(int32_t, Protomux *)> ready,
std::function<void(int32_t)> gone);
// ---- Stream consumer: the connection WITHOUT our Protomux -----------------
//
// For an embedder that runs its own protocol stack over the secret-stream —
// keetd puts protomux 3.12.0 + hypercore replication there. OPT-IN: when no
// consumer is set, nothing below changes and every connection gets the
// internal Protomux + godot-rpc channel exactly as before.
//
// When a consumer IS set, a connection that reaches noise_ready gets NO
// internal Protomux and no godot-rpc channel. Instead, all on the UDX thread:
// open(info) once, at noise_ready
// frame(id, p, n) every non-empty decrypted secret-stream message
// (keepalives are still eaten here, as the reference does)
// drain(id) the stream fell back under its write watermark
// close(id) the stream closed
// Set it BEFORE starting: a connection that already got the internal mux keeps
// it. Streams already open without a mux are replayed to `open`.
struct StreamInfo {
int32_t peer_id;
bool is_initiator; // the Noise initiator — secret-stream's isInitiator
uint8_t handshake_hash[64]; // secret-stream's handshakeHash
uint8_t remote_public_key[32]; // secret-stream's remotePublicKey
};
void 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);
// UDX THREAD ONLY (from a consumer callback, or run_on_loop). Encrypts and
// writes one message; returns false once the stream is past its watermark
// (wait for drain) or the peer is gone.
bool send_frame(int32_t peer_id, const uint8_t *data, size_t len);
// UDX THREAD ONLY. Destroys the stream; `close` follows.
void close_stream(int32_t peer_id);
std::function<void(const StreamInfo &)> _stream_open;
std::function<void(int32_t, const uint8_t *, size_t)> _stream_frame;
std::function<void(int32_t)> _stream_drain;
std::function<void(int32_t)> _stream_close;
// Internal, but declared here rather than after a `private:` — the insertion
// point sits inside a public section and closing it would silently privatise
// every declaration below, which is exactly what happened the first time.
std::deque<std::function<void()>> _tasks;
std::mutex _tasks_mutex;
// The async handle is memset to zero in the constructor and only initialised
// inside _thread_func, ON the loop thread. Signalling it before that
// dereferences a null loop and takes the process down — which is exactly what
// happened, because a consumer polls from frame one while the thread is still
// starting. Tasks queued early are safe; they are drained by the tick timer
// once the loop is up.
// The UDX loop's clock, sampled each tick. Everything on that thread that
// needs "now" reads this rather than calling a clock of its own.
uint64_t _loop_now_ms = 0;
std::atomic<bool> _async_ready{ false };
// Signal the loop. The ONLY way — see the comment on the definition.
void _wake();
// Called from Godot _poll() to fire peer signals on main thread.
bool pop_peer_event(PeerEvent &out);
// Called from Godot _poll() to fire dht_peer_found signals.
bool pop_peer_found_event(PeerFoundEvent &out);
// Called from Godot _poll() to fire DHT storage signals.
bool pop_storage_event(DhtStorageEvent &out);
// Callbacks for peer events (set before start).
PeerConnectedCallback on_peer_connected;
PeerDisconnectedCallback on_peer_disconnected;
// ---- DHT discovery API ----
// Start the DHT stack with Ed25519 keypair and bootstrap nodes.
// Must be called BEFORE start_server/start_client or after to use DHT discovery.
// The bootstrap list this transport will use — its OWN copy, readable for a
// test that wants to know two peers did not end up sharing one.
const std::vector<std::string> &bootstrap_nodes() const { return _bootstrap; }
// The public HyperDHT bootstrap nodes. ONE DEFINITION, here, because the
// duplicate that used to live only in hyperswarm_peer.cpp meant "no bootstrap
// given" behaved one way for Godot and another for every other embedder.
static const char *const DEFAULT_BOOTSTRAP[3];
// ⛔ AN EMPTY LIST MEANS THE PUBLIC DHT, NOT AN ISOLATED ONE. Passing nothing
// used to leave _bootstrap_count at 0, so bootstrap() sent zero FIND_NODEs and
// on_bootstrapped never fired — a transport that is "active" and permanently
// unreachable, with no error anywhere. The session-daemon shipped that way:
// two Blenders on one topic, both reporting joined, neither ever on the
// network. Callers that want isolation pass a testnet address, which every
// gate here already does — and is why none of them caught this.
void start_dht(const Ed25519KeyPair &kp,
const char **bootstrap_nodes, int bootstrap_count);
// Connect to a remote peer via DHT (by Ed25519 public key).
void dht_connect(const uint8_t *remote_public_key);
// Start listening for incoming DHT connections.
void dht_listen();
// Announce a topic on the DHT.
void dht_announce(const uint8_t *topic);
// Lookup a topic on the DHT.
void dht_lookup(const uint8_t *topic);
// Join a topic with server and/or client mode (multi-session, auto-refresh).
void dht_join_topic(const uint8_t *topic, bool server, bool client);
// Leave a topic (decrements session, unannounces when all server sessions gone).
void dht_leave_topic(const uint8_t *topic);
// ---- DHT Storage API ----
// Store an immutable value on the DHT. key = BLAKE2b(value).
// Result arrives via pop_storage_event (IMMUTABLE_PUT_COMPLETE).
void dht_immutable_put(const uint8_t *value, size_t value_len);
// Retrieve an immutable value from the DHT by key.
// Result arrives via pop_storage_event (IMMUTABLE_GET_COMPLETE).
void dht_immutable_get(const uint8_t *key);
// Store a mutable value on the DHT signed with our keypair.
// Result arrives via pop_storage_event (MUTABLE_PUT_COMPLETE).
void dht_mutable_put(const uint8_t *value, size_t value_len, uint64_t seq);
// Retrieve a mutable value from the DHT by public key.
// Result arrives via pop_storage_event (MUTABLE_GET_COMPLETE).
void dht_mutable_get(const uint8_t *public_key, uint64_t min_seq);
// ---- Swarm Manager API ----
// High-level swarm join/leave (auto-connect, retry, priority queue).
void swarm_join(const uint8_t *topic, bool server, bool client);
void swarm_leave(const uint8_t *topic);
void swarm_join_peer(const uint8_t *pk);
void swarm_leave_peer(const uint8_t *pk);
// Pop swarm events for main thread.
bool pop_swarm_event(SwarmEvent &out);
// Get swarm status snapshot (thread-safe via copy).
SwarmStatus get_swarm_status() const;
// Get a connected peer's public key (32 bytes). Returns false if peer not found.
bool get_peer_public_key(int32_t peer_id, uint8_t *out_pk) const;
// Get a connected peer's RTT stats (ms). Returns false if peer not found.
struct PeerRttStats {
uint32_t srtt_ms; // Smoothed RTT
uint32_t rttvar_ms; // RTT variance
uint32_t rto_ms; // Retransmission timeout
uint32_t min_rtt_ms; // Minimum RTT (BBR)
};
bool get_peer_rtt(int32_t peer_id, PeerRttStats &out) const;
bool is_dht_active() const { return _dht_active.load(); }
bool is_dht_bootstrapped() const { return _dht_bootstrapped.load(); }
bool is_running() const { return _running.load(); }
int bound_port() const { return _bound_port; }
// Get punch statistics (returns nullptr if DHT not active).
const PunchStats *punch_stats() const {
return _dht_connector ? &_dht_connector->punch_stats() : nullptr;
}
// Get a snapshot of all active DHT connections for diagnostics.
// Returns the number of entries written. Thread-safe (reads atomic fields).
int get_connection_diagnostics(DhtConnectionInfo *out, int max_count) const {
if (!_dht_connector) return 0;
uint64_t now = uv_hrtime() / 1000000ULL; // nanoseconds → milliseconds
return _dht_connector->get_connections_info(out, max_count, now);
}
private:
// Background thread entry point.
void _thread_func();
// libudx callbacks (static, forward to instance via user data).
static void _on_stream_read(udx_stream_t *stream, ssize_t read_len, const uv_buf_t *buf);
static void _on_stream_recv(udx_stream_t *stream, ssize_t read_len, const uv_buf_t *buf);
static void _on_stream_close(udx_stream_t *stream, int status);
static void _on_stream_finalize(udx_stream_t *stream);
static void _on_socket_recv(udx_socket_t *socket, ssize_t read_len,
const uv_buf_t *buf, const struct sockaddr *from);
static void _on_write_ack(udx_stream_write_t *req, int status, int unordered);
static void _on_stream_drain(udx_stream_t *stream);
static void _on_send_complete(udx_stream_send_t *req, int status);
// Internal peer management (called on UDX thread only).
PeerConnection *_get_peer(int32_t peer_id);
PeerConnection *_add_peer(int32_t peer_id, uint32_t local_sid, uint32_t remote_sid,
const struct sockaddr *addr,
udx_socket_t *socket_override = nullptr);
void _remove_peer(int32_t peer_id);
int32_t _next_peer_id();
// Dispatch an incoming packet (decrypt if needed, route Noise messages).
void _dispatch_incoming(PeerConnection *peer, const uint8_t *buf_data,
ssize_t buf_len, PacketTransferMode mode_override);
// Handle an incoming Noise handshake/header message on the UDX thread.
void _handle_noise_message(PeerConnection *peer, NoiseMessageType msg_type,
const uint8_t *payload, size_t payload_len);
// Send raw bytes on a peer's UDX stream (bypassing encryption).
void _send_raw(PeerConnection *peer, const uint8_t *data, size_t len);
// Encrypt and send via UDX stream (used as Protomux write callback).
void _send_encrypted(PeerConnection *peer, const uint8_t *data, size_t len);
// Route decrypted payload to Godot PacketQueue (used by godot-rpc channel).
void _deliver_godot_packet(PeerConnection *peer, const uint8_t *data, size_t len);
// Handle DHT-discovered connection ready to become a UDX stream.
void _dht_on_connected(uint32_t conn_id, const DhtConnection &conn);
// Initialize DHT stack on the UDX thread.
void _init_dht();
// Command queue (main thread -> UDX thread).
enum CommandType {
CMD_SEND,
CMD_DISCONNECT,
CMD_CONNECT_TO,
CMD_DHT_INIT,
CMD_DHT_CONNECT,
CMD_DHT_LISTEN,
CMD_DHT_ANNOUNCE,
CMD_DHT_LOOKUP,
CMD_DHT_JOIN_TOPIC,
CMD_DHT_LEAVE_TOPIC,
CMD_DHT_IMMUTABLE_PUT,
CMD_DHT_IMMUTABLE_GET,
CMD_DHT_MUTABLE_PUT,
CMD_DHT_MUTABLE_GET,
CMD_SWARM_JOIN,
CMD_SWARM_LEAVE,
CMD_SWARM_JOIN_PEER,
CMD_SWARM_LEAVE_PEER,
};
struct Command {
CommandType type;
int32_t peer_id;
std::vector<uint8_t> data;
PacketTransferMode mode;
int32_t channel;
std::string address;
int port;
bool force;
bool server; // For join_topic.
bool client; // For join_topic.
uint64_t seq; // For mutable storage operations.
uint8_t remote_public_key[32]; // For encrypted connections.
};
std::deque<Command> _commands;
std::mutex _cmd_mutex;
void _queue_command(Command &&cmd);
// State
uv_loop_t _loop;
udx_t _udx;
udx_socket_t _socket;
uv_async_t _async_handle;
uv_timer_t _tick_timer;
std::thread _thread;
std::atomic<bool> _running;
std::atomic<bool> _is_server;
int _bound_port;
std::map<int32_t, PeerConnection *> _peers;
std::map<std::string, int32_t> _peers_by_pubkey; // hex(remote_pk) → peer_id for dedup
std::mutex _peers_mutex;
uint32_t _next_stream_id;
int32_t _peer_id_counter;
// Encryption state.
bool _encrypted;
NoiseKeypair _local_keypair;
uint8_t _remote_public_key[32]; // Expected remote key (client mode).
PacketQueue _incoming;
std::deque<PeerEvent> _peer_events;
std::deque<PeerFoundEvent> _peer_found_events;
std::deque<DhtStorageEvent> _storage_events;
std::deque<SwarmEvent> _swarm_events;
std::mutex _events_mutex;
// DHT state (owned by UDX thread).
std::atomic<bool> _dht_active;
std::atomic<bool> _dht_bootstrapped;
DhtRpc *_dht_rpc;
HyperDht *_hyperdht;
DhtConnector *_dht_connector;
Ed25519KeyPair _dht_keypair;
// OWNED, because the caller's strings are gone by the time these are read.
// start_dht only QUEUES the init; the parse happens later, on the UDX thread,
// in _init_dht. Holding the caller's `const char **` meant every peer needed
// its bootstrap argument to stay alive and unmoved across a thread boundary
// for an unbounded time — see start_dht.
std::vector<std::string> _bootstrap;
// Swarm manager (owned by UDX thread, null until DHT init).
SwarmManager *_swarm_manager;
};
#endif // HYPERSWARM_UDX_TRANSPORT_H