// swarmd-session — session-daemon's socket, swarmd's insides.
//
// swarmd-session --topic <seed> --socket <path> [--bootstrap host:port]
//
// Argument for argument, and byte for byte on both sides, this is
// hyperswarm-peer/daemon/session-daemon: the topic is BLAKE2b(seed), the local
// socket carries opaque length-prefixed frames, and the swarm channel is
// "blender-bridge-session/v1" keyed by the topic — so it PAIRS WITH AN EXISTING
// session-daemon PEER and does not force a flag day.
//
// ⛔ WHY REPLACE SOMETHING THAT WORKS. session_daemon.cpp fans out to local
// clients from the UDX thread with unchecked writes on non-blocking fds, and
// says so in its own header: a partial write there does not drop a frame, it
// DESYNCHRONISES the client's stream, and the client then reads a garbage length
// and closes. It has no backpressure, no client ids, no liveness, and no content
// plane. swarmd has all of those and they are gated. Keeping two transports for
// one job means fixing everything twice.
//
// ⛔ AND IT WRITES NO WIRE OF ITS OWN. The legacy socket is swarmd::UdsServer,
// which already speaks exactly this framing with the per-client buffering
// session-daemon lacks; the swarm side is swarmd::Client. This file is the
// hundred lines of plumbing between them, and that is all it should ever be.
#include <atomic>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <csignal>
#include <string>
#include <thread>
#include <vector>
#include <unistd.h>
#include <sodium.h>
#include "client.h"
#include "daemon.h"
#include "uds_server.h"
static const char *SESSION_PROTOCOL = "blender-bridge-session/v1";
// ⛔ A DAEMON MUST NAME ITS OWN PROCESS. Without this the only way to stop one
// is to match its command line, and a pattern that matches a daemon's argv also
// matches the shell that typed it — `pkill -f` killed an ssh session's own
// remote shell twice in this project. The pidfile sits beside the socket, so
// whoever can reach the socket can stop exactly the process behind it.
static void write_pidfile(const std::string &socket_path) {
FILE *f = fopen((socket_path + ".pid").c_str(), "w");
if (!f) return; // diagnostics, never a blocker
fprintf(f, "%ld\n", (long)getpid());
fclose(f);
}
static void remove_pidfile(const std::string &socket_path) {
::unlink((socket_path + ".pid").c_str());
}
static volatile sig_atomic_t g_stop = 0;
static void on_signal(int) { g_stop = 1; }
int main(int argc, char **argv) {
if (sodium_init() < 0) { fprintf(stderr, "sodium_init failed\n"); return 1; }
const char *seed = nullptr, *bootstrap = nullptr, *sock = nullptr;
const char *storage = nullptr;
for (int i = 1; i < argc; i++) {
if (!strcmp(argv[i], "--topic") && i + 1 < argc) seed = argv[++i];
else if (!strcmp(argv[i], "--bootstrap") && i + 1 < argc) bootstrap = argv[++i];
else if (!strcmp(argv[i], "--socket") && i + 1 < argc) sock = argv[++i];
else if (!strcmp(argv[i], "--storage") && i + 1 < argc) storage = argv[++i];
}
if (!seed || !sock) {
fprintf(stderr, "usage: swarmd-session --topic <seed> --socket <path> "
"[--bootstrap host:port] [--storage dir]\n");
return 1;
}
uint8_t topic[32];
crypto_generichash(topic, sizeof(topic), (const uint8_t *)seed, strlen(seed),
nullptr, 0);
signal(SIGINT, on_signal);
signal(SIGTERM, on_signal);
// The swarmd daemon lives beside the legacy socket rather than at a fixed
// path: two sessions on one machine must not collide, and the caller already
// chose a unique name for the socket it asked for.
const std::string swarmd_sock = std::string(sock) + ".swarmd";
::unlink(swarmd_sock.c_str());
swarmd::Daemon daemon;
if (!daemon.start(swarmd_sock.c_str(), bootstrap, storage)) {
fprintf(stderr, "failed to start swarmd on %s\n", swarmd_sock.c_str());
return 1;
}
std::atomic<bool> stop_daemon(false);
std::thread daemon_thread([&daemon, &stop_daemon]() {
while (!stop_daemon.load()) daemon.poll(5);
});
swarmd::Client client;
if (!client.connect(swarmd_sock, "swarmd-session")) {
fprintf(stderr, "cannot reach the swarmd we just started: %s\n",
client.last_error().c_str());
stop_daemon.store(true);
daemon_thread.join();
daemon.stop();
return 1;
}
if (!client.join(topic)) {
fprintf(stderr, "join failed: %s\n", client.last_error().c_str());
return 1;
}
// ⛔ THE CHANNEL ID IS THE TOPIC, as session_daemon.cpp:93 has it
// (`mux->create_channel(PROTOCOL, _topic)`). Protomux pairs on protocol AND
// id, so getting this wrong produces two peers that mesh perfectly and never
// exchange a frame.
const std::vector<uint8_t> chan_id(topic, topic + 32);
const uint64_t channel = client.open_channel(SESSION_PROTOCOL, chan_id, 1);
if (channel == 0) {
fprintf(stderr, "channel failed: %s\n", client.last_error().c_str());
return 1;
}
swarmd::UdsServer legacy;
::unlink(sock);
if (!legacy.listen(sock)) {
fprintf(stderr, "cannot listen on %s\n", sock);
return 1;
}
uint64_t to_peers = 0, to_local = 0;
// A frame from a peer goes to every local client.
client.on.on_message = [&](uint64_t, uint64_t, uint64_t, uint64_t,
const uint8_t *bytes, size_t len) {
legacy.broadcast(bytes, len);
to_local++;
};
client.on.on_disconnect = [](const std::string &why) {
fprintf(stderr, "[swarmd-session] lost the daemon: %s\n", why.c_str());
};
// A frame from a local client goes to the peers AND to the other locals —
// session-daemon fans out both ways (session_daemon.cpp:189) and a second
// local client on one machine would otherwise never hear the first.
legacy.on_frame = [&](uint64_t from, const uint8_t *data, size_t len) {
std::vector<uint8_t> copy(data, data + len);
legacy.broadcast(copy.data(), copy.size(), from);
// `accepted == false` means no peer has the channel open yet. That is
// normal before the swarm meshes and is not an error to report per frame.
if (client.send(channel, 0, 0, copy.data(), copy.size())) to_peers++;
};
// The pidfile goes with the CONTROL socket, because that is the one
// swarmd-ctl is pointed at: whatever socket you can reach, socket + ".pid"
// names the process behind it.
write_pidfile(swarmd_sock);
printf("[swarmd-session] listening on %s (swarmd at %s, pid %ld)\n",
sock, swarmd_sock.c_str(), (long)getpid());
fflush(stdout);
// ⛔ SAY WHETHER THE SWARM IS REACHABLE, not just whether we started. "Bound"
// is a local fact; two peers once sat on one topic for ten minutes both
// reporting success while neither had ever bootstrapped.
bool said_up = false;
int ticks = 0;
while (!g_stop) {
legacy.poll(5);
if (!client.poll(5)) break;
if ((ticks++ % 200) == 0) {
swarmd::Status st;
if (client.status(st)) {
if (!said_up && st.bootstrapped) {
printf("[swarmd-session] dht=up\n");
said_up = true;
}
printf("[swarmd-session] dht=%s peers=%llu local=%d "
"to_peers=%llu to_local=%llu\n",
st.bootstrapped ? "up" : "connecting",
(unsigned long long)st.peers_connected, legacy.client_count(),
(unsigned long long)to_peers, (unsigned long long)to_local);
fflush(stdout);
}
}
}
printf("[swarmd-session] stopping\n");
legacy.close_all();
client.close();
stop_daemon.store(true);
daemon_thread.join();
daemon.stop();
::unlink(sock);
::unlink(swarmd_sock.c_str());
remove_pidfile(swarmd_sock);
return 0;
}