#ifndef HYPERSWARM_HYPERDHT_H
#define HYPERSWARM_HYPERDHT_H
// HyperDHT layer — announce/lookup/find_peer on top of DHT-RPC.
// Wire-compatible with holepunchto/hyperdht.
// Implements Noise IK handshake relay and peer discovery.
#include "compact_encoding.h"
#include "dht_messages.h"
#include "dht_rpc.h"
#include "dht_storage.h"
#include "noise_ik.h"
#include <cstdint>
#include <cstring>
#include <functional>
#include <string>
#include <unordered_map>
#include <vector>
// ---- Constants ----
static constexpr int HYPERDHT_MAX_QUERIES = 16;
static constexpr int HYPERDHT_MAX_PEERS = 64;
// Topic refresh timing — matches JS peer-discovery.js.
// ⛔ THE SERVER'S OWN RECORD MUST BE KEPT ALIVE, and it was announced once.
// listen() announced BLAKE2b(our public key) and nothing ever refreshed it: the
// self-announce is not a TopicDiscovery, so _tick_topics never saw it. DHT
// records expire, and when ours did, find_peer(our_pk) found nothing and the
// peer was unreachable until it restarted.
//
// JS gives this its own component — hyperdht's Announcer (lib/announcer.js),
// constructed per SERVER against hash(keyPair.publicKey) (server.js:173-175),
// distinct from hyperswarm's topic-level PeerDiscovery. Its _background loop is
// 100 iterations of a 3s sleep, so a full re-announce every ~5 minutes.
static constexpr uint64_t SELF_ANNOUNCE_UPDATE_MS = 300000; // 5 min, = JS 100 x 3s.
// ⛔ AND CHECK THE HOLDERS ARE STILL THERE. Re-announcing on a timer alone means
// losing every node that holds our record costs a full interval before anyone
// notices. JS pings each holder every 3s and re-announces the moment fewer than
// MIN_ACTIVE answer (announcer.js:104-125) — 3 seconds versus 5 minutes to
// notice we have become unreachable.
static constexpr uint64_t SELF_ANNOUNCE_PING_MS = 3000;
// Tag carried on holder pings so their replies can be told apart from every
// other PING this node makes.
static constexpr uint32_t HYPERDHT_TAG_SELF_PING = 0x5E1F0001;
static constexpr int SELF_ANNOUNCE_MIN_ACTIVE = 3; // JS MIN_ACTIVE
static constexpr uint64_t TOPIC_REFRESH_INTERVAL_MS = 600000; // 10 minutes.
static constexpr uint64_t TOPIC_REFRESH_JITTER_MS = 120000; // 0–2 minutes random.
static constexpr uint64_t TOPIC_REFRESH_GRACE_MS = 30000; // 30s grace (laptop wake).
// ⛔ EARLY RETRY WHILE A TOPIC HAS FOUND NOBODY. A deliberate deviation from JS,
// for a case JS's cadence does not serve.
//
// join_topic announces then chains ONE lookup. If that lookup runs before the
// other peer's announce has reached the nodes it queries, it returns nothing —
// and the next attempt is REFRESH_INTERVAL away. Ten to twelve minutes of a
// session that says "joined" and will never connect. Measured on the public DHT:
// two daemons on one topic meshed in 1 of 3 attempts, and no amount of waiting
// under ten minutes changed the other 2.
//
// JS gets away with it because its asymmetric case is the common one — a
// long-lived server announced minutes ago, and the joining client's first lookup
// finds a record that is already settled. Two peers joining a session within
// seconds of each other is the symmetric case, and BOTH sides can lose the race
// at once.
//
// So: back off from 2s while nobody has been found, cap at a minute, and hand
// the topic to the JS cadence the moment it finds anyone.
static constexpr uint64_t TOPIC_DISCOVERY_RETRY_MS = 2000; // first retry.
static constexpr uint64_t TOPIC_DISCOVERY_RETRY_MAX_MS = 60000; // ceiling.
static constexpr int TOPIC_DISCOVERY_RETRY_SHIFTS = 5; // 2s..64s -> capped.
// ---- Found peer from lookup/find_peer ----
struct FoundPeer {
uint8_t public_key[32];
IPv4Address relay_addresses[8];
int relay_count;
IPv4Address relay_node; // The DHT node that stored this peer's record (from field).
};
// ---- Pending handshake (incoming or outgoing) ----
struct DhtHandshake {
uint32_t id;
NoiseIK *noise;
bool is_initiator;
uint8_t remote_public_key[32];
IPv4Address relay_address; // DHT node relaying the handshake.
uint32_t firewall;
// After handshake completion.
uint8_t holepunch_secret[32];
uint8_t tx_key[32];
uint8_t rx_key[32];
uint8_t hash[64];
bool complete;
bool active;
// Server address from the handshake reply (relay's peerAddress field).
// This is the server's DHT node address as seen by the relay.
IPv4Address server_address;
bool has_server_address;
// Client address (for responder): the initiator's address as seen by the relay.
// Set from FROM_RELAY peerAddress so the responder can connect back.
IPv4Address client_address;
bool has_client_address;
// Remote's NoisePayload (decoded from handshake).
dht_msg::NoisePayload remote_payload;
bool has_remote_payload;
// RESPONDER DEDUP (hyperdht server.js _onpeerhandshake: `_connects.get(hex(noise))`).
// A repeated copy of the same inbound handshake — a retried request, or the same one
// relayed by a second node — gets the cached reply of the first instead of a new
// handshake, stream id and connection. Keyed by BLAKE2b-256 of the Noise message 1;
// `reply_noise` is the Noise message 2 sent for it, OWNED (freed in _free_handshake).
uint8_t noise_key[32];
bool has_noise_key;
std::vector<uint8_t> *reply_noise;
// ⛔ WHEN THIS SLOT WAS TAKEN. Without it nothing could ever reclaim a slot,
// the 16-slot pool filled with handshakes that had already served their
// purpose, and _handle_handshake answered every later peer with a bare
// `return` — no reply, no log. See test/test_handshake_pool.cpp and
// docs/PARITY-hyperswarm.md §8. JS reclaims the same way (server.js
// _clearLater / handshakeClearWait) plus on raw-stream close.
uint64_t created_at_ms;
};
// ---- Topic discovery session ----
// Matches JS Hyperswarm's PeerDiscovery/PeerDiscoverySession pattern.
// Multiple sessions can share a single topic discovery; topic is cleaned
// up only when all sessions have left (reference counting).
static constexpr int TOPIC_MAX_SESSIONS = 8;
static constexpr int TOPIC_MAX_DISCOVERIES = 32;
struct TopicDiscovery {
uint8_t topic[32];
int server_sessions; // Number of active sessions acting as server.
int client_sessions; // Number of active sessions acting as client.
uint64_t last_refresh; // Timestamp of last announce/lookup.
uint64_t refresh_at; // Next scheduled refresh (with jitter).
uint32_t current_query_id; // In-flight query (0 = none).
bool active;
bool needs_unannounce; // True if we announced and need to unannounce on leave.
bool pending_lookup; // Lookup queued after announce completes (server+client).
// ⛔ HAS THIS TOPIC EVER FOUND ANYONE? A topic that has is on the JS 10-minute
// cadence, which is right: the announce persists in the DHT and re-querying
// sooner buys nothing. A topic that has NOT is in a race it may already have
// lost — see _schedule_topic_refresh.
int peers_seen;
int empty_refreshes; // Consecutive refreshes that found nobody (backoff).
};
// ---- HyperDht class ----
class HyperDht {
public:
HyperDht();
~HyperDht();
// Initialize with a DhtRpc instance and our Ed25519 keypair.
void init(DhtRpc *rpc, const Ed25519KeyPair &keypair);
// Start a LOOKUP query for a topic (32-byte hash).
// Returns query ID. Results arrive via on_peer_found callback.
uint32_t lookup(const uint8_t *topic);
// Start an ANNOUNCE for a topic. Signs with our Ed25519 key.
// Returns query ID.
uint32_t announce(const uint8_t *topic);
// Unannounce from a topic.
void unannounce(const uint8_t *topic);
// Start a FIND_PEER query for a specific peer's public key.
uint32_t find_peer(const uint8_t *public_key);
// Initiate a handshake to a remote peer through a relay node.
// Returns handshake ID.
uint32_t connect(const uint8_t *remote_public_key,
const IPv4Address &relay);
// Start listening for incoming handshakes.
void listen();
// When the self-announce is next due, or 0 when not listening. For gates:
// a record that is never refreshed is invisible from the outside until it
// expires, which is why this is observable at all.
uint64_t self_announce_due_at() const { return _self_announce_at; }
uint64_t self_announce_count() const { return _self_announce_count; }
uint64_t self_ping_rounds() const { return _self_ping_rounds; }
int self_holder_count() const;
// How many recorded holders carry a NON-ZERO token. The routing-table guess
// this replaced wrote `memset(token, 0, 32)`, so a zero token is the
// signature of the old behaviour and the only thing that tells the two
// apart from outside. See PARITY-hyperswarm.md §15.
int self_holder_tokens() const;
// Process time-based events.
void tick(uint64_t now_ms);
// Handshakes currently held (initiator + responder). For gates.
int handshake_count() const { return _handshake_count; }
// ---- Topic management (matches JS Hyperswarm join/leave) ----
// Join a topic with server and/or client mode.
// Increments session count. If first session, creates discovery and
// immediately announces (if server) or looks up (if client).
// Returns true on success.
bool join_topic(const uint8_t *topic, bool server, bool client);
// Leave a topic. Decrements session count. If server sessions drop
// to 0 and topic was announced, unannounces. If all sessions are 0,
// removes the discovery entirely.
void leave_topic(const uint8_t *topic);
// Get the number of active topic discoveries.
int topic_count() const { return _topic_count; }
// Find a topic discovery by topic hash.
const TopicDiscovery *find_topic(const uint8_t *topic) const;
// Get the topic associated with a query ID. Returns nullptr if not found.
const uint8_t *get_query_topic(uint32_t query_id) const;
// ---- DHT Storage API ----
// Store an immutable value on the DHT.
// target = BLAKE2b(value). Queries closest nodes, then stores.
// Result arrives via on_immutable_put_complete callback.
void immutable_put(const uint8_t *value, size_t value_len);
// Retrieve an immutable value from the DHT by its key (BLAKE2b hash).
// Result arrives via on_immutable_get_complete callback.
void immutable_get(const uint8_t *key);
// Store a mutable value on the DHT signed with our keypair.
// target = BLAKE2b(publicKey). Queries closest nodes, then stores with signature.
// Result arrives via on_mutable_put_complete callback.
void 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 on_mutable_get_complete callback.
void mutable_get(const uint8_t *public_key, uint64_t min_seq = 0);
// Access the local storage cache.
DhtStorage &storage() { return _storage; }
const DhtStorage &storage() const { return _storage; }
// Get our Ed25519 public key.
const uint8_t *public_key() const { return _keypair.public_key; }
// Access the underlying DhtRpc (used by DhtConnector for sending).
DhtRpc *rpc() const { return _rpc; }
// Access namespaces.
const uint8_t *ns_peer_handshake() const { return _ns_peer_handshake; }
const uint8_t *ns_peer_holepunch() const { return _ns_peer_holepunch; }
const uint8_t *ns_announce() const { return _ns_announce; }
// ---- Callbacks ----
// Called when a peer is found during lookup/find_peer.
std::function<void(uint32_t query_id, const FoundPeer &peer)> on_peer_found;
// Called when a query (lookup/announce/find_peer) completes.
std::function<void(uint32_t query_id)> on_query_complete;
// Called when an incoming handshake is received (if listening).
std::function<void(uint32_t handshake_id, const uint8_t *remote_pk,
const dht_msg::NoisePayload &payload)> on_handshake_request;
// Called when an outgoing handshake completes.
std::function<void(uint32_t handshake_id, const DhtHandshake &hs)> on_handshake_complete;
// ---- Storage callbacks ----
// Called when immutable_put completes. Provides the 32-byte key.
std::function<void(const uint8_t *key)> on_immutable_put_complete;
// Called when immutable_get completes. value=nullptr if not found.
std::function<void(const uint8_t *key, const uint8_t *value, size_t value_len)>
on_immutable_get_complete;
// Called when mutable_put completes. success=true if stored.
std::function<void(bool success)> on_mutable_put_complete;
// Called when mutable_get completes. value=nullptr if not found.
std::function<void(const uint8_t *public_key, const uint8_t *value,
size_t value_len, uint64_t seq)> on_mutable_get_complete;
// ---- Holepunch callbacks ----
// Called when an incoming PEER_HOLEPUNCH request is received (from relay).
std::function<void(const DhtRequest &req)> on_holepunch_request;
// Called when a holepunch RPC response arrives (reply to our outgoing request).
// tag is 0x10000 | conn_id.
std::function<void(uint32_t tag, const uint8_t *value, size_t value_len,
const IPv4Address &from)> on_holepunch_response;
private:
// Handle DHT-RPC responses routed to us.
void _on_rpc_response(const DhtResponse &res, uint32_t tag);
// Handle incoming DHT-RPC requests (PEER_HANDSHAKE, PEER_HOLEPUNCH).
void _on_rpc_request(const DhtRequest &req);
// Handle an incoming handshake message.
void _handle_handshake(const DhtRequest &req);
// Decode lookup reply value and extract peers.
void _process_lookup_value(uint32_t query_id, const uint8_t *value, size_t value_len,
const IPv4Address &from);
// Decode find_peer reply value (single peer record) and extract peer.
void _process_find_peer_value(uint32_t query_id, const uint8_t *value, size_t value_len,
const IPv4Address &from);
// ⛔ ONE SIGNABLE FOR ANNOUNCE AND UNANNOUNCE. They used to be built by two
// separate functions and they DIVERGED: announce carried the correct
// ns || generichash_batch(...), unannounce carried ns || topic || peer, so
// JS rejected every unannounce we ever sent and our records never withdrew.
// JS has one function for both (persistent.js annSignable), which is why
// theirs could not drift. Public for the vector gate.
// out64[0..32) = ns
// out64[32..64) = generichash_batch([target, node_id, token, peer, refresh])
public:
static void ann_signable(uint8_t *out64, const uint8_t *ns,
const uint8_t *target, const uint8_t *node_id,
const uint8_t *token,
const uint8_t *peer_buf, size_t peer_len);
private:
// Per-node (un)announce value. `ns` selects _ns_announce or _ns_unannounce;
// everything else is identical, which is the point.
std::vector<uint8_t> _encode_ann_value_for_node(
const uint8_t *ns, const uint8_t *topic, const uint8_t *token,
const uint8_t *node_id);
std::vector<uint8_t> _encode_announce_value_for_node(
const uint8_t *topic, const uint8_t *token, const uint8_t *node_id);
// Sign an announce record.
void _sign_announce(const uint8_t *signable, size_t signable_len, uint8_t *signature);
// Record a node that ACCEPTED our announce, with the token it issued.
// Called from the commit callback, which is the only place the holder's
// address exists. See PARITY-hyperswarm.md §15.
void _note_announce_holder(const uint8_t *topic, const IPv4Address &from,
const uint8_t *token);
// Handshake management.
DhtHandshake *_alloc_handshake(bool initiator, const uint8_t *remote_pk,
const IPv4Address &relay);
DhtHandshake *_find_handshake(uint32_t id);
void _free_handshake(DhtHandshake *hs);
// Reclaim handshakes that have outlived HANDSHAKE_TIMEOUT_MS. Called from
// tick(). This is the ONLY thing that returns a slot for a handshake that
// SUCCEEDED — _free_handshake's other call site is a decrypt failure.
void _sweep_handshakes(uint64_t now_ms);
// tick()'s clock, remembered so _alloc_handshake can stamp with the SAME
// time base the sweep compares against. A slot stamped from one clock and
// swept against another is a bug waiting for a test to drive tick() with a
// synthetic value — and a fix that cannot be driven on demand cannot be
// proven. 0 until the first tick; the sweep adopts those (see the .cpp).
uint64_t _last_tick_ms = 0;
// Self-announce (see SELF_ANNOUNCE_UPDATE_MS).
uint8_t _self_target[32] = { 0 };
uint64_t _self_announce_at = 0; // 0 = not listening
uint64_t _self_announce_count = 0;
uint64_t _self_ping_at = 0;
uint64_t _self_ping_rounds = 0;
int _self_ping_sent = 0; // holders pinged in the round now in flight
int _self_ping_alive = 0; // how many answered
void _tick_self_announce(uint64_t now_ms);
void _ping_self_holders(uint64_t now_ms);
void _finalize_handshake(DhtHandshake *hs);
// Respond to any non-internal query with closer nodes + token + optional value.
// Matches JS dht-rpc behavior where every node responds to queries.
void _reply_to_query(const DhtRequest &req);
// Handle incoming storage requests (IMMUTABLE_PUT/GET, MUTABLE_PUT/GET).
void _handle_storage_request(const DhtRequest &req);
// Handle storage query responses (for immutable/mutable get).
void _handle_storage_response(const DhtResponse &res, uint32_t tag);
// State.
DhtRpc *_rpc;
Ed25519KeyPair _keypair;
bool _listening;
// Namespace hashes.
uint8_t _ns_announce[32];
uint8_t _ns_unannounce[32];
uint8_t _ns_mutable_put[32];
uint8_t _ns_peer_handshake[32];
uint8_t _ns_peer_holepunch[32];
// Active handshakes.
// The pool, and how long an entry may sit in it. A handshake that has not
// produced a connection within this window is abandoned by definition: the
// far side either punched through (and no longer needs us to hold state) or
// never will.
static constexpr int MAX_HANDSHAKES = 16;
static constexpr uint64_t HANDSHAKE_TIMEOUT_MS = 30000;
DhtHandshake _handshakes[MAX_HANDSHAKES];
int _handshake_count;
uint32_t _handshake_id_counter;
// DHT local storage.
DhtStorage _storage;
// Track query → topic mapping for announce commit.
struct QueryInfo {
uint32_t query_id;
uint8_t topic[32];
bool is_announce;
bool is_lookup;
bool is_find_peer;
bool is_immutable_get;
bool is_immutable_put;
bool is_mutable_get;
bool is_mutable_put;
uint8_t storage_key[32]; // For immutable: BLAKE2b(value); for mutable: publicKey.
uint64_t mutable_seq;
std::vector<uint8_t> put_value; // Value to store (for put operations).
uint8_t put_signature[64]; // Signature (for mutable put).
bool active;
};
QueryInfo _query_infos[HYPERDHT_MAX_QUERIES];
int _query_info_count;
// Track announced topics for proper unannounce (closest nodes + tokens).
struct AnnouncedNode {
IPv4Address address;
uint8_t token[32];
};
struct AnnouncedTopic {
uint8_t topic[32];
AnnouncedNode nodes[20];
int node_count;
};
// Key: hex string of topic.
std::unordered_map<std::string, AnnouncedTopic> _announced_topics;
// ---- Topic discovery management ----
// Matches JS Hyperswarm's PeerDiscovery with session counting.
TopicDiscovery _topics[TOPIC_MAX_DISCOVERIES];
int _topic_count;
TopicDiscovery *_find_topic_mut(const uint8_t *topic);
TopicDiscovery *_alloc_topic(const uint8_t *topic);
void _free_topic(TopicDiscovery *td);
void _refresh_topic(TopicDiscovery *td, uint64_t now_ms);
void _note_peer_for_query(uint32_t query_id);
void _schedule_topic_refresh(TopicDiscovery *td, uint64_t now_ms, bool eager);
void _tick_topics(uint64_t now_ms);
uint64_t _random_jitter() const;
static std::string _topic_hex(const uint8_t *topic);
};
#endif // HYPERSWARM_HYPERDHT_H