#include "swarm_manager.h"
#include "dht_connector.h"
#include "hyperdht.h"
#include <algorithm>
#include <cstdio>
#include <cstdlib>
// ---- SwarmBulkTimer ----
// Exact match of JS lib/bulk-timer.js.
SwarmBulkTimer::SwarmBulkTimer()
: _interval_ms(1000), _next_fire(0), _started(false) {
}
void SwarmBulkTimer::set_time(uint64_t time_ms) {
// JS: setInterval(fn, Math.floor(this._time * 0.66))
_interval_ms = (time_ms * 66) / 100;
if (_interval_ms < 1) _interval_ms = 1;
}
void SwarmBulkTimer::add(SwarmPeerInfo *info) {
// JS: if (!this._interval) { this._interval = setInterval(...) }
// We use tick-driven instead of setInterval, so just start tracking.
if (!_started) {
// Will set _next_fire on next tick() call.
_started = false; // Defer to tick().
}
_pending.push_back(info);
}
void SwarmBulkTimer::tick(uint64_t now_ms, std::vector<SwarmPeerInfo *> &flushed) {
if (_pending.empty() && _next.empty()) {
_started = false;
return;
}
if (!_started) {
_next_fire = now_ms + _interval_ms;
_started = true;
}
if (now_ms >= _next_fire) {
// JS: _ontick()
// if (this._next.length) this._fn(this._next)
if (!_next.empty()) {
for (auto *info : _next) {
flushed.push_back(info);
}
}
// this._next = this._pending; this._pending = []
_next.clear();
_next.swap(_pending);
_next_fire = now_ms + _interval_ms;
if (_next.empty() && _pending.empty()) {
_started = false;
}
}
}
bool SwarmBulkTimer::empty() const {
return _pending.empty() && _next.empty();
}
// ---- SwarmRetryTimer ----
// Exact match of JS lib/retry-timer.js.
SwarmRetryTimer::SwarmRetryTimer() {
}
void SwarmRetryTimer::init(uint64_t instance_jitter) {
// JS module-level backoff computation:
// BACKOFF_S = 1000 + round(500 * random())
// BACKOFF_M = 5000 + round(1000 * random())
// BACKOFF_L = 15000 + round(2000 * random())
// BACKOFF_X = 600000 + round(120000 * random())
uint64_t backoff_s = SWARM_BACKOFF_S_BASE + (rand() % (SWARM_BACKOFF_S_JITTER + 1));
uint64_t backoff_m = SWARM_BACKOFF_M_BASE + (rand() % (SWARM_BACKOFF_M_JITTER + 1));
uint64_t backoff_l = SWARM_BACKOFF_L_BASE + (rand() % (SWARM_BACKOFF_L_JITTER + 1));
uint64_t backoff_x = SWARM_BACKOFF_X_BASE + (rand() % (SWARM_BACKOFF_X_JITTER + 1));
// JS per-instance: new BulkTimer(backoffs[i] + round(jitter * random()), push)
uint64_t j_s = (instance_jitter > 0) ? (rand() % (instance_jitter + 1)) : 0;
uint64_t j_m = (instance_jitter > 0) ? (rand() % (instance_jitter + 1)) : 0;
uint64_t j_l = (instance_jitter > 0) ? (rand() % (instance_jitter + 1)) : 0;
uint64_t j_x = (instance_jitter > 0) ? (rand() % (instance_jitter + 1)) : 0;
_timer_s.set_time(backoff_s + j_s);
_timer_m.set_time(backoff_m + j_m);
_timer_l.set_time(backoff_l + j_l);
_timer_x.set_time(backoff_x + j_x);
}
bool SwarmRetryTimer::add(SwarmPeerInfo *info) {
SwarmBulkTimer *timer = _select_timer(info);
if (!timer) return false;
timer->add(info);
return true;
}
void SwarmRetryTimer::tick(uint64_t now_ms, std::vector<SwarmPeerInfo *> &flushed) {
_timer_s.tick(now_ms, flushed);
_timer_m.tick(now_ms, flushed);
_timer_l.tick(now_ms, flushed);
_timer_x.tick(now_ms, flushed);
}
// JS: _selectRetryTimer(peerInfo) — exact match
SwarmBulkTimer *SwarmRetryTimer::_select_timer(SwarmPeerInfo *info) {
if (info->banned || !info->reconnecting) return nullptr;
if (info->attempts > 3) {
return info->is_explicit ? &_timer_x : nullptr;
}
if (info->attempts == 0) return &_timer_s;
if (info->proven) {
switch (info->attempts) {
case 1: return &_timer_s;
case 2: return &_timer_m;
case 3: return &_timer_l;
}
} else {
switch (info->attempts) {
case 1: return &_timer_m;
case 2: return &_timer_l;
case 3: return &_timer_l;
}
}
return nullptr;
}
// ---- SwarmManager ----
SwarmManager::SwarmManager()
: _max_peers(SWARM_MAX_PEERS), _max_parallel(SWARM_MAX_PARALLEL),
_connecting(0), _all_connections(0), _client_connections(0),
_connector(nullptr), _dht(nullptr),
_now_ms(0), _total_connections(0), _draining_queue(false) {
memset(_local_pk, 0, 32);
}
SwarmManager::~SwarmManager() {
for (auto &pair : _peers) {
delete pair.second;
}
_peers.clear();
}
void SwarmManager::init(DhtConnector *connector, HyperDht *dht,
const uint8_t *local_pk) {
_connector = connector;
_dht = dht;
memcpy(_local_pk, local_pk, 32);
_retry.init(SWARM_INSTANCE_JITTER);
}
void SwarmManager::tick(uint64_t now_ms) {
_now_ms = now_ms;
// Tick retry timers — flushed peers go through _requeue.
std::vector<SwarmPeerInfo *> flushed;
_retry.tick(now_ms, flushed);
if (!flushed.empty()) {
_requeue(flushed);
}
}
// ---- API methods ----
void SwarmManager::join(const uint8_t *topic, bool server, bool client) {
std::string topic_hex = _pk_to_hex(topic);
_topics[topic_hex] = {server, client};
}
void SwarmManager::leave(const uint8_t *topic) {
std::string topic_hex = _pk_to_hex(topic);
_topics.erase(topic_hex);
}
// JS: joinPeer(publicKey)
void SwarmManager::join_peer(const uint8_t *pk) {
SwarmPeerInfo *info = _upsert_peer(pk);
if (!info) return; // self
if (_explicit_peers.find(info) == _explicit_peers.end()) {
info->is_explicit = true;
_explicit_peers.insert(info);
}
// JS: if (this._allConnections.has(publicKey)) return
// Skip if already connected or connect in progress.
if (info->peer_id != 0) return;
if (info->connecting) return;
if (info->update_priority()) {
_enqueue(info);
}
}
// JS: leavePeer(publicKey)
void SwarmManager::leave_peer(const uint8_t *pk) {
std::string hex = _pk_to_hex(pk);
auto it = _peers.find(hex);
if (it == _peers.end()) return;
SwarmPeerInfo *info = it->second;
info->is_explicit = false;
_explicit_peers.erase(info);
_maybe_delete_peer(info);
}
// ---- DHT callbacks ----
// JS: _handlePeer(peer, topic) — exact match of logic
void SwarmManager::on_peer_found(const uint8_t *pk) {
on_peer_found(pk, nullptr);
}
void SwarmManager::on_peer_found(const uint8_t *pk, const uint8_t *topic) {
// Only auto-connect if SwarmManager has active topics or the peer is explicit.
// Prevents interference when old join_topic/connect_to_peer API is used.
if (_topics.empty()) {
std::string hex = _pk_to_hex(pk);
auto it = _peers.find(hex);
if (it == _peers.end() || !it->second->is_explicit) {
return;
}
}
SwarmPeerInfo *info = _upsert_peer(pk);
if (!info) return;
// ⛔ ASSOCIATE THE TOPIC BEFORE THE EARLY RETURNS, as JS does.
//
// JS _handlePeer (hyperswarm/index.js:449) is:
//
// const peerInfo = this._upsertPeer(...)
// if (peerInfo) peerInfo._topic(topic) // here
// if (!peerInfo || this._allConnections.has(...)) return // then this
//
// This used to run AFTER the already-connected and connecting checks, so a
// peer we were already talking to that then turned up on a SECOND topic
// never recorded it. _should_requeue decides reconnection by walking exactly
// that list, so a peer found on A and B, connected via A, and dropped after
// A was left would be requeued by JS and abandoned by us.
if (topic) {
info->add_topic(_pk_to_hex(topic));
} else if (!_topics.empty()) {
// No specific topic — associate every active one, which is the same
// information a per-topic discovery would eventually have supplied.
for (const auto &pair : _topics) {
info->add_topic(pair.first);
}
}
// JS: if (!peerInfo || this._allConnections.has(peer.publicKey)) return
// _allConnections tracks peers from connect() call (before handshake).
if (info->peer_id != 0) return;
if (info->connecting) return;
// JS: if (!peerInfo.prioritized || peerInfo.server) peerInfo._reset()
if (!info->prioritized() || info->server()) {
info->reset();
}
// JS: if (peerInfo._updatePriority()) { this._enqueue(peerInfo) }
if (info->update_priority()) {
_enqueue(info);
}
}
// Called when DhtConnector completes a connection.
void SwarmManager::on_peer_connected(int32_t peer_id, const uint8_t *pk, bool is_initiator) {
std::string hex = _pk_to_hex(pk);
SwarmPeerInfo *info;
auto it = _peers.find(hex);
if (it != _peers.end()) {
info = it->second;
} else {
// Server-side connection: peer connected to us, we didn't discover them.
info = _upsert_peer(pk);
if (!info) return; // self (shouldn't happen)
}
// If a server connection arrives for a peer with a pending client connection,
// cancel the client-side counters. The client attempt will eventually call
// on_connect_failed() but info->connecting will be false, so it'll bail out.
bool had_pending_client = info->connecting && !is_initiator;
// JS: peerInfo._connected() — matches JS conn.on('open') handler.
info->connecting = false;
info->on_connected(_now_ms);
info->client = is_initiator;
info->peer_id = peer_id;
info->queued = false;
_total_connections++;
if (is_initiator) {
// Client connection completed — counter was already incremented in _connect().
// Call _connect_done() to decrement _connecting and try more connections.
_connect_done();
} else {
// Server connection — add to _allConnections.
_all_connections++;
if (had_pending_client) {
// Clean up leaked client connection counters.
// _connect() had incremented: _connecting, _all_connections, _client_connections.
// None of those were decremented since we're getting a server connection instead.
if (_connecting > 0) _connecting--;
if (_all_connections > 0) _all_connections--;
if (_client_connections > 0) _client_connections--;
}
}
// Track reverse mapping for disconnect lookup.
_peers_by_id[peer_id] = hex;
// Push event to main thread.
SwarmEvent evt;
evt.peer_id = peer_id;
memcpy(evt.public_key, pk, 32);
evt.is_client = is_initiator;
evt.connected = true;
_events.push_back(evt);
}
void SwarmManager::on_peer_disconnected(int32_t peer_id) {
auto id_it = _peers_by_id.find(peer_id);
if (id_it == _peers_by_id.end()) return;
std::string hex = id_it->second;
_peers_by_id.erase(id_it);
auto it = _peers.find(hex);
if (it == _peers.end()) return;
SwarmPeerInfo *info = it->second;
bool was_client = info->client;
// JS: peerInfo._disconnected()
info->on_disconnected(_now_ms);
info->peer_id = 0;
_total_connections--;
_all_connections--;
// JS: this._clientConnections-- (only for client-initiated connections)
if (was_client) {
_client_connections--;
}
// Push disconnect event.
SwarmEvent evt;
evt.peer_id = peer_id;
memcpy(evt.public_key, info->public_key, 32);
evt.is_client = was_client;
evt.connected = false;
_events.push_back(evt);
// JS: peerInfo.waiting = this._shouldRequeue(peerInfo) && this._timer.add(peerInfo)
info->waiting = _should_requeue(info) && _retry.add(info);
_maybe_delete_peer(info);
// JS: this._attemptClientConnections()
_attempt_client_connections();
}
// Called when a connect attempt fails before a stream is established.
void SwarmManager::on_connect_failed(const uint8_t *pk) {
std::string hex = _pk_to_hex(pk);
auto it = _peers.find(hex);
if (it == _peers.end()) {
// Unknown peer — still need to decrement if we had a connecting slot.
if (_connecting > 0) _connecting--;
return;
}
SwarmPeerInfo *info = it->second;
// Guard: if on_peer_connected() already processed this peer,
// info->connecting is false and _connectDone() was already called.
// Don't double-decrement. (JS: conn.on('close') checks !opened)
if (!info->connecting) return;
info->connecting = false;
_all_connections--;
_client_connections--;
_connect_done();
// JS: peerInfo._disconnected() is called even on failed connects.
info->on_disconnected(_now_ms);
// JS: peerInfo.waiting = this._shouldRequeue(peerInfo) && this._timer.add(peerInfo)
info->waiting = _should_requeue(info) && _retry.add(info);
_maybe_delete_peer(info);
}
// ---- Event queue ----
bool SwarmManager::pop_event(SwarmEvent &out) {
if (_events.empty()) return false;
out = _events.front();
_events.pop_front();
return true;
}
// ---- Diagnostics ----
SwarmStatus SwarmManager::get_status() const {
SwarmStatus s;
s.peers_known = static_cast<int>(_peers.size());
s.peers_connected = _total_connections;
s.connecting = _connecting;
s.queue_size = static_cast<int>(_queue.size());
s.max_peers = _max_peers;
s.max_parallel = _max_parallel;
s.explicit_count = static_cast<int>(_explicit_peers.size());
s.all_connections = _all_connections;
s.client_connections = _client_connections;
// JS: flush() resolves when connecting===0 && queue empty && _withinMaxPeers()
s.flushed = (_connecting == 0) && _queue.empty();
return s;
}
// ---- Internal helpers ----
// JS: _shouldConnect() — exact match (minus destroyed/suspended which we don't have)
bool SwarmManager::_should_connect() const {
return _connecting < _max_parallel && _within_max_peers();
}
// JS: _shouldConnectExplicit() — explicit peers bypass _withinMaxPeers
bool SwarmManager::_should_connect_explicit() const {
return _connecting < _max_parallel;
}
// JS: _withinMaxPeers() — exact match
// JS uses _allConnections.size which counts BOTH pending client + established connections.
bool SwarmManager::_within_max_peers() const {
int e = static_cast<int>(_explicit_peers.size());
int factor;
if (e == 0) factor = 1;
else if (e < 2) factor = 2;
else if (e < 4) factor = 3;
else factor = 4;
return factor * _all_connections < _max_peers;
}
// JS: _shouldRequeue(peerInfo) — exact match
bool SwarmManager::_should_requeue(SwarmPeerInfo *info) const {
if (info->is_explicit) return true;
// Check if any of the peer's topics are still being discovered.
for (const auto &topic : info->topics) {
if (_topics.find(topic) != _topics.end()) {
return true;
}
}
return false;
}
// JS: _enqueue(peerInfo) — exact match
void SwarmManager::_enqueue(SwarmPeerInfo *info) {
if (info->queued) return;
info->queued = true;
// Insert sorted: highest priority first, FIFO within same priority.
auto pos = _queue.begin();
while (pos != _queue.end() && (*pos)->priority >= info->priority) {
++pos;
}
_queue.insert(pos, info);
_attempt_client_connections();
}
// JS: _requeue(batch) — BulkTimer callback, exact match
void SwarmManager::_requeue(std::vector<SwarmPeerInfo *> &batch) {
for (auto *info : batch) {
info->waiting = false;
// JS: if (peerInfo._updatePriority() === false ||
// this._allConnections.has(peerInfo.publicKey) || peerInfo.queued)
if (!info->update_priority()) continue;
if (info->peer_id != 0) continue; // already connected
if (info->connecting) continue; // connect in progress (JS: _allConnections.has)
if (info->queued) continue;
info->queued = true;
// Insert sorted.
auto pos = _queue.begin();
while (pos != _queue.end() && (*pos)->priority >= info->priority) {
++pos;
}
_queue.insert(pos, info);
}
_attempt_client_connections();
}
// JS: _attemptClientConnections() — exact match
void SwarmManager::_attempt_client_connections() {
if (_draining_queue) return;
_draining_queue = true;
// First pass: explicit peers (JS: for (const peerInfo of this.explicitPeers))
for (auto *info : _explicit_peers) {
if (!_should_connect_explicit()) break;
// JS: if (peerInfo.attempts >= 5 || Date.now() - peerInfo.disconnectedTime < peerInfo.attempts * 1000)
if (info->attempts >= 5) continue;
if (_now_ms - info->disconnected_at < (uint64_t)info->attempts * 1000) continue;
if (info->peer_id != 0) continue; // already connected
if (info->banned) continue;
_connect(info);
}
// Second pass: drain priority queue.
// JS: while (this._queue.length && this._shouldConnect())
while (!_queue.empty() && _should_connect()) {
SwarmPeerInfo *info = _queue.front();
_queue.pop_front();
info->queued = false;
_connect(info);
}
_draining_queue = false;
}
// JS: _connect(peerInfo, queued) — initiate connection via DhtConnector
void SwarmManager::_connect(SwarmPeerInfo *info) {
// JS: if (peerInfo.banned || this._allConnections.has(peerInfo.publicKey)) return
if (info->banned) return;
if (info->peer_id != 0) return; // already connected
if (info->connecting) return; // connect already in progress
info->tried = true;
info->connecting = true; // Matches JS: _allConnections.add(conn) at connect time.
_connecting++;
_all_connections++; // JS: _allConnections.add(conn)
_client_connections++; // JS: this._clientConnections++
_connector->connect(info->public_key);
}
// JS: _connectDone()
void SwarmManager::_connect_done() {
if (_connecting <= 0) return; // Guard: never go negative.
_connecting--;
if (_connecting < _max_parallel) {
_attempt_client_connections();
}
}
// JS: _upsertPeer(publicKey)
SwarmPeerInfo *SwarmManager::_upsert_peer(const uint8_t *pk) {
// JS: if (b4a.equals(publicKey, this.keyPair.publicKey)) return null
if (memcmp(pk, _local_pk, 32) == 0) return nullptr;
std::string hex = _pk_to_hex(pk);
auto it = _peers.find(hex);
if (it != _peers.end()) return it->second;
SwarmPeerInfo *info = new SwarmPeerInfo();
memcpy(info->public_key, pk, 32);
_peers[hex] = info;
return info;
}
// JS: _maybeDeletePeer(peerInfo)
void SwarmManager::_maybe_delete_peer(SwarmPeerInfo *info) {
if (!info->should_gc()) return;
if (info->peer_id != 0) return; // has active connection
if (info->connecting) return; // JS: _allConnections.has(peerInfo.publicKey)
std::string hex = _pk_to_hex(info->public_key);
auto it = _peers.find(hex);
if (it != _peers.end() && it->second == info) {
_peers.erase(it);
delete info;
}
}
std::string SwarmManager::_pk_to_hex(const uint8_t *pk) {
char buf[65];
for (int i = 0; i < 32; i++) {
snprintf(buf + i * 2, 3, "%02x", pk[i]);
}
buf[64] = '\0';
return std::string(buf, 64);
}