#include "daemon.h"
#include <cstdio>
#include <cstdlib>
#include <cstring>
namespace swarmd {
// liboverlay names content in hex; the wire carries 32 raw bytes.
static std::string _hex32(const uint8_t h[32]) {
static const char *d = "0123456789abcdef";
std::string out(64, '0');
for (int i = 0; i < 32; i++) { out[i * 2] = d[h[i] >> 4]; out[i * 2 + 1] = d[h[i] & 15]; }
return out;
}
bool Daemon::start(const char *socket_path, const char *bootstrap,
const char *storage_dir) {
_socket_path = socket_path;
if (!_uds.listen(socket_path)) return false;
ControlBackend backend;
backend.status = [this](swarmd_status_t *out) { _fill_status(out); };
backend.swarm_join = [this](const uint8_t topic[32], bool server, bool client) {
_t.swarm_join(topic, server, client);
return true;
};
backend.swarm_leave = [this](const uint8_t topic[32]) {
_t.swarm_leave(topic);
return true;
};
backend.channel_open = [this](const std::string &protocol,
const std::vector<uint8_t> &id,
uint32_t slots, uint64_t owner) -> uint64_t {
const uint64_t handle = _next_channel++;
// Registered HERE, not on the loop: a client that sends the instant
// channel-open returns must get "open on nobody yet", not "no such
// handle". Those are different answers and only one of them means retry.
auto open_peers = std::make_shared<std::atomic<int>>(0);
{
std::lock_guard<std::mutex> lk(_chan_m);
_chan_open[handle] = open_peers;
}
// ⛔ PROTOMUX BELONGS TO THE UDX THREAD. Creating a channel from here
// would touch its handles from the wrong thread; run_on_loop is the only
// way in, and it is why this returns the handle immediately rather than
// waiting for the channel to exist.
_t.run_on_loop([this, handle, protocol, id, slots, owner, open_peers]() {
ChannelSpec &spec = _channels[handle];
spec.handle = handle;
spec.protocol = protocol;
spec.id = id;
spec.slots = slots ? slots : 1;
spec.owner = owner;
spec.open_peers = open_peers;
for (auto &kv : _muxes) _open_on_peer(spec, kv.first, kv.second);
_publish_open(handle);
});
return handle;
};
backend.channel_close = [this](uint64_t handle) {
{
std::lock_guard<std::mutex> lk(_chan_m);
_chan_open.erase(handle);
}
_t.run_on_loop([this, handle]() {
auto it = _channels.find(handle);
if (it == _channels.end()) return;
for (auto &kv : it->second.per_peer) {
if (kv.second) kv.second->close();
}
_channels.erase(it);
});
return true;
};
backend.send = [this](uint64_t handle, uint32_t slot, uint64_t target,
const uint8_t *data, size_t len, bool *writable) {
// ⛔ DO NOT ANSWER ABOUT THE QUEUE. This used to return true and set
// writable=true unconditionally, which meant "I posted a lambda" — while
// ProtomuxChannel::send drops the frame outright when the channel is not
// yet open, and returns false to nobody. A caller told "accepted" stops
// looking. That is #72's wall in miniature: the failure path was silent.
//
// What is knowable here is whether the handle exists and whether the
// channel is open on any peer at all. A TARGETED send can still miss
// (the aggregate does not say which peer), and that case is reported by
// the drop counter below rather than pretended away.
int open = -1;
{
std::lock_guard<std::mutex> lk(_chan_m);
auto it = _chan_open.find(handle);
if (it != _chan_open.end()) open = it->second->load(std::memory_order_relaxed);
}
if (open <= 0) {
*writable = false;
return false;
}
std::vector<uint8_t> copy(data, data + len); // the frame buffer is reused
*writable = true;
_t.run_on_loop([this, handle, slot, target, copy]() {
auto it = _channels.find(handle);
if (it == _channels.end()) return;
for (auto &kv : it->second.per_peer) {
if (target != 0 && (uint64_t)(uint32_t)kv.first != target) continue;
if (kv.second == nullptr) continue;
if (!kv.second->send(slot, copy.data(), copy.size())) {
// The channel closed between the answer and the write. Say so
// once per drop rather than let the frame disappear.
_dropped_sends++;
fprintf(stderr,
"[swarmd] channel %llu: protomux refused a %zu-byte frame "
"to peer %d (dropped=%llu)\n",
(unsigned long long)handle, copy.size(), kv.first,
(unsigned long long)_dropped_sends);
}
}
});
return true;
};
// ---- content ----
// hash_of_file touches no overlay state, so it answers on the calling thread.
// The other three are orders that cross to the UDX loop inside Content.
backend.content_hash = [this](const std::string &path, uint8_t out[32]) {
const std::string hex = Content::hash_of_file(path);
if (hex.size() != 64) return false;
for (int i = 0; i < 32; i++)
out[i] = (uint8_t)strtoul(hex.substr(i * 2, 2).c_str(), nullptr, 16);
return true;
};
backend.content_serve = [this](const uint8_t hash[32], const std::string &path) {
if (!_content.is_open()) return false;
_content.serve(_hex32(hash), path);
return true;
};
backend.content_fetch = [this](const uint8_t hash[32], const std::string &dest) {
if (!_content.is_open()) return false;
_content.fetch(_hex32(hash), dest);
return true;
};
backend.content_cancel = [this](const uint8_t hash[32], bool discard) {
if (!_content.is_open()) return false;
_content.cancel(_hex32(hash), discard);
return true;
};
backend.daemon_pubkey = [this](uint8_t out[32]) { memcpy(out, _pubkey, 32); };
_control.attach(&_uds, backend);
// Peer lifecycle arrives on the UDX thread. Enqueue only.
_t.set_mux_callbacks(
[this](int32_t id, Protomux *mux) {
// UDX THREAD. Record the mux and give this peer every channel a
// client has already opened — a channel opened before the swarm
// meshed must still reach peers that arrive later.
_muxes[id] = mux;
for (auto &kv : _channels) _open_on_peer(kv.second, id, mux);
// swarmd owns the peer lifecycle, so it builds the overlay
// channel itself: set_mux_callbacks holds ONE pair of callbacks
// and whoever sets it last owns them.
_content.add_peer(id, mux);
Event e;
e.kind = Event::PEER_UP;
e.peer_id = id;
memset(e.pubkey, 0, 32);
_t.get_peer_public_key(id, e.pubkey);
_push(e);
},
[this](int32_t id) {
// UDX THREAD. The mux is going away; drop our references to its
// channels rather than let them dangle.
_muxes.erase(id);
_content.remove_peer(id);
for (auto &kv : _channels) {
kv.second.per_peer.erase(id);
_publish_open(kv.first);
}
Event e;
e.kind = Event::PEER_DOWN;
e.peer_id = id;
memset(e.pubkey, 0, 32);
_push(e);
});
Ed25519KeyPair kp = NoiseIK::generate_keypair();
memcpy(_pubkey, kp.public_key, 32);
if (bootstrap) {
const char *bs[1] = { bootstrap };
_t.start_dht(kp, bs, 1);
} else {
// ⛔ NO BOOTSTRAP MEANS THE PUBLIC DHT, not an isolated one. Passing an
// empty list used to leave the transport "active" and permanently
// unreachable with no error anywhere (hyper-games #68 bug A).
_t.start_dht(kp, nullptr, 0);
}
// AFTER start_dht: Content marshals onto the UDX loop, which has to exist.
// A daemon with no storage dir simply has no content plane; its commands are
// refused above rather than silently accepted.
if (storage_dir && *storage_dir) {
if (!_content.open(storage_dir, &_t))
fprintf(stderr, "[swarmd] content plane is OFF: %s is not usable\n", storage_dir);
}
_started = true;
return true;
}
void Daemon::stop() {
if (!_started) return;
_started = false;
_t.set_mux_callbacks(nullptr, nullptr);
_content.close();
_t.stop();
_uds.close_all();
}
void Daemon::_open_on_peer(ChannelSpec &spec, int32_t peer_id, Protomux *mux) {
// UDX THREAD ONLY.
if (!mux) return;
if (spec.per_peer.count(peer_id)) return;
ProtomuxChannel *ch = mux->create_channel(spec.protocol, spec.id);
if (ch == nullptr) return; // already open on this mux
const uint64_t handle = spec.handle;
// ⛔ `open()` IS NOT OPEN. ProtomuxChannel::opened turns true in
// _fully_open(), i.e. only once the REMOTE has opened the same channel —
// and until then ProtomuxChannel::send returns false and discards. So the
// only trustworthy source for "can I write" is protomux's own callbacks.
ch->on_open = [this, handle](const uint8_t *, size_t) { _publish_open(handle); };
ch->on_close = [this, handle, peer_id](bool) {
auto it = _channels.find(handle);
if (it != _channels.end()) it->second.per_peer.erase(peer_id);
_publish_open(handle);
};
for (uint32_t slot = 0; slot < spec.slots; slot++) {
ch->add_message([this, handle, slot, peer_id](const uint8_t *data, size_t len,
ProtomuxChannel *) {
// UDX THREAD. Enqueue only — the fds belong to the main thread.
Event e;
e.kind = Event::MESSAGE;
e.peer_id = peer_id;
memset(e.pubkey, 0, 32);
e.channel = handle;
e.slot = slot;
e.bytes.assign(data, data + len);
_push(e);
});
}
spec.per_peer[peer_id] = ch;
ch->open();
_publish_open(handle);
}
void Daemon::_publish_open(uint64_t handle) {
// UDX THREAD ONLY.
auto it = _channels.find(handle);
if (it == _channels.end()) return;
if (!it->second.open_peers) return;
int n = 0;
for (auto &kv : it->second.per_peer) {
if (kv.second && kv.second->opened && !kv.second->closed) n++;
}
it->second.open_peers->store(n, std::memory_order_relaxed);
}
void Daemon::_push(const Event &e) {
std::lock_guard<std::mutex> lk(_m);
if (_events.size() >= MAX_EVENTS) {
_dropped_events++;
return;
}
_events.push_back(e);
}
void Daemon::_drain() {
std::deque<Event> batch;
{
std::lock_guard<std::mutex> lk(_m);
batch.swap(_events);
}
for (const Event &e : batch) {
uint8_t *out = nullptr;
size_t out_len = 0;
if (e.kind == Event::MESSAGE) {
swarmd_message_t m;
memset(&m, 0, sizeof(m));
m.channel = e.channel;
m.slot = e.slot;
m.origin_kind = 0; // 0 = remote peer, 1 = another local client
m.origin_id = (uint64_t)(uint32_t)e.peer_id;
m.bytes = (uint8_t *)e.bytes.data();
m.bytes_len = e.bytes.size();
if (swarmd_encode_on_message(&m, &out, &out_len) < 0) continue;
} else if (e.kind == Event::PEER_UP) {
swarmd_peer_up_t up;
memset(&up, 0, sizeof(up));
up.peer = (uint64_t)(uint32_t)e.peer_id;
memcpy(up.pubkey, e.pubkey, 32);
if (swarmd_encode_on_peer_up(&up, &out, &out_len) < 0) continue;
} else {
swarmd_peer_down_t down;
memset(&down, 0, sizeof(down));
down.peer = (uint64_t)(uint32_t)e.peer_id;
if (swarmd_encode_on_peer_down(&down, &out, &out_len) < 0) continue;
}
if (out == nullptr) continue;
_uds.broadcast(out, out_len);
free(out); // caller frees
}
}
void Daemon::_fill_status(swarmd_status_t *out) {
memset(out, 0, sizeof(*out));
out->dht_state = _t.is_dht_bootstrapped() ? 2 : (_t.is_dht_active() ? 1 : 0);
out->bootstrapped = _t.is_dht_bootstrapped();
const SwarmStatus st = _t.get_swarm_status();
out->peers_known = (uint64_t)st.peers_known;
out->peers_connected = (uint64_t)st.peers_connected;
out->connecting = (uint64_t)st.connecting;
out->queued = (uint64_t)st.queue_size;
out->local_clients = (uint64_t)_uds.client_count();
}
int Daemon::peers() const {
return const_cast<UdxTransport &>(_t).get_swarm_status().peers_connected;
}
void Daemon::_drain_content() {
if (!_content.is_open()) return;
std::vector<Content::Event> batch;
_content.poll(batch);
for (const Content::Event &e : batch) {
uint8_t raw[32];
memset(raw, 0, 32);
for (int i = 0; i < 32 && (size_t)(i * 2 + 1) < e.hash.size(); i++)
raw[i] = (uint8_t)strtoul(e.hash.substr(i * 2, 2).c_str(), nullptr, 16);
uint8_t *out = nullptr;
size_t out_len = 0;
if (e.kind == Content::Event::DONE) {
swarmd_content_done_t d;
memset(&d, 0, sizeof(d));
memcpy(d.hash, raw, 32);
d.path.data = (utf8_t *)e.detail.c_str();
d.path.len = e.detail.size();
if (swarmd_encode_on_content_done(&d, &out, &out_len) < 0) continue;
} else if (e.kind == Content::Event::FAILED) {
swarmd_content_failed_t f;
memset(&f, 0, sizeof(f));
memcpy(f.hash, raw, 32);
f.reason.data = (utf8_t *)e.detail.c_str();
f.reason.len = e.detail.size();
if (swarmd_encode_on_content_failed(&f, &out, &out_len) < 0) continue;
} else {
swarmd_content_progress_t p;
memset(&p, 0, sizeof(p));
memcpy(p.hash, raw, 32);
p.received = e.received;
p.total = e.total;
if (swarmd_encode_on_content_progress(&p, &out, &out_len) < 0) continue;
}
if (out == nullptr) continue;
_uds.broadcast(out, out_len);
free(out); // caller frees
}
}
void Daemon::poll(int timeout_ms) {
_drain();
_drain_content();
_uds.poll(timeout_ms);
}
} // namespace swarmd