| … | … | @@ -0,0 +1,180 @@ |
| 1 | // swarmd-session — session-daemon's socket, swarmd's insides. |
| 2 | // |
| 3 | // swarmd-session --topic <seed> --socket <path> [--bootstrap host:port] |
| 4 | // |
| 5 | // Argument for argument, and byte for byte on both sides, this is |
| 6 | // hyperswarm-peer/daemon/session-daemon: the topic is BLAKE2b(seed), the local |
| 7 | // socket carries opaque length-prefixed frames, and the swarm channel is |
| 8 | // "blender-bridge-session/v1" keyed by the topic — so it PAIRS WITH AN EXISTING |
| 9 | // session-daemon PEER and does not force a flag day. |
| 10 | // |
| 11 | // ⛔ WHY REPLACE SOMETHING THAT WORKS. session_daemon.cpp fans out to local |
| 12 | // clients from the UDX thread with unchecked writes on non-blocking fds, and |
| 13 | // says so in its own header: a partial write there does not drop a frame, it |
| 14 | // DESYNCHRONISES the client's stream, and the client then reads a garbage length |
| 15 | // and closes. It has no backpressure, no client ids, no liveness, and no content |
| 16 | // plane. swarmd has all of those and they are gated. Keeping two transports for |
| 17 | // one job means fixing everything twice. |
| 18 | // |
| 19 | // ⛔ AND IT WRITES NO WIRE OF ITS OWN. The legacy socket is swarmd::UdsServer, |
| 20 | // which already speaks exactly this framing with the per-client buffering |
| 21 | // session-daemon lacks; the swarm side is swarmd::Client. This file is the |
| 22 | // hundred lines of plumbing between them, and that is all it should ever be. |
| 23 | |
| 24 | #include <atomic> |
| 25 | #include <cstdio> |
| 26 | #include <cstdlib> |
| 27 | #include <cstring> |
| 28 | #include <csignal> |
| 29 | #include <string> |
| 30 | #include <thread> |
| 31 | #include <vector> |
| 32 | |
| 33 | #include <unistd.h> |
| 34 | |
| 35 | #include <sodium.h> |
| 36 | |
| 37 | #include "client.h" |
| 38 | #include "daemon.h" |
| 39 | #include "uds_server.h" |
| 40 | |
| 41 | static const char *SESSION_PROTOCOL = "blender-bridge-session/v1"; |
| 42 | |
| 43 | static volatile sig_atomic_t g_stop = 0; |
| 44 | static void on_signal(int) { g_stop = 1; } |
| 45 | |
| 46 | int main(int argc, char **argv) { |
| 47 | if (sodium_init() < 0) { fprintf(stderr, "sodium_init failed\n"); return 1; } |
| 48 | |
| 49 | const char *seed = nullptr, *bootstrap = nullptr, *sock = nullptr; |
| 50 | const char *storage = nullptr; |
| 51 | for (int i = 1; i < argc; i++) { |
| 52 | if (!strcmp(argv[i], "--topic") && i + 1 < argc) seed = argv[++i]; |
| 53 | else if (!strcmp(argv[i], "--bootstrap") && i + 1 < argc) bootstrap = argv[++i]; |
| 54 | else if (!strcmp(argv[i], "--socket") && i + 1 < argc) sock = argv[++i]; |
| 55 | else if (!strcmp(argv[i], "--storage") && i + 1 < argc) storage = argv[++i]; |
| 56 | } |
| 57 | if (!seed || !sock) { |
| 58 | fprintf(stderr, "usage: swarmd-session --topic <seed> --socket <path> " |
| 59 | "[--bootstrap host:port] [--storage dir]\n"); |
| 60 | return 1; |
| 61 | } |
| 62 | |
| 63 | uint8_t topic[32]; |
| 64 | crypto_generichash(topic, sizeof(topic), (const uint8_t *)seed, strlen(seed), |
| 65 | nullptr, 0); |
| 66 | |
| 67 | signal(SIGINT, on_signal); |
| 68 | signal(SIGTERM, on_signal); |
| 69 | |
| 70 | // The swarmd daemon lives beside the legacy socket rather than at a fixed |
| 71 | // path: two sessions on one machine must not collide, and the caller already |
| 72 | // chose a unique name for the socket it asked for. |
| 73 | const std::string swarmd_sock = std::string(sock) + ".swarmd"; |
| 74 | ::unlink(swarmd_sock.c_str()); |
| 75 | |
| 76 | swarmd::Daemon daemon; |
| 77 | if (!daemon.start(swarmd_sock.c_str(), bootstrap, storage)) { |
| 78 | fprintf(stderr, "failed to start swarmd on %s\n", swarmd_sock.c_str()); |
| 79 | return 1; |
| 80 | } |
| 81 | |
| 82 | std::atomic<bool> stop_daemon(false); |
| 83 | std::thread daemon_thread([&daemon, &stop_daemon]() { |
| 84 | while (!stop_daemon.load()) daemon.poll(5); |
| 85 | }); |
| 86 | |
| 87 | swarmd::Client client; |
| 88 | if (!client.connect(swarmd_sock, "swarmd-session")) { |
| 89 | fprintf(stderr, "cannot reach the swarmd we just started: %s\n", |
| 90 | client.last_error().c_str()); |
| 91 | stop_daemon.store(true); |
| 92 | daemon_thread.join(); |
| 93 | daemon.stop(); |
| 94 | return 1; |
| 95 | } |
| 96 | if (!client.join(topic)) { |
| 97 | fprintf(stderr, "join failed: %s\n", client.last_error().c_str()); |
| 98 | return 1; |
| 99 | } |
| 100 | |
| 101 | // ⛔ THE CHANNEL ID IS THE TOPIC, as session_daemon.cpp:93 has it |
| 102 | // (`mux->create_channel(PROTOCOL, _topic)`). Protomux pairs on protocol AND |
| 103 | // id, so getting this wrong produces two peers that mesh perfectly and never |
| 104 | // exchange a frame. |
| 105 | const std::vector<uint8_t> chan_id(topic, topic + 32); |
| 106 | const uint64_t channel = client.open_channel(SESSION_PROTOCOL, chan_id, 1); |
| 107 | if (channel == 0) { |
| 108 | fprintf(stderr, "channel failed: %s\n", client.last_error().c_str()); |
| 109 | return 1; |
| 110 | } |
| 111 | |
| 112 | swarmd::UdsServer legacy; |
| 113 | ::unlink(sock); |
| 114 | if (!legacy.listen(sock)) { |
| 115 | fprintf(stderr, "cannot listen on %s\n", sock); |
| 116 | return 1; |
| 117 | } |
| 118 | |
| 119 | uint64_t to_peers = 0, to_local = 0; |
| 120 | |
| 121 | // A frame from a peer goes to every local client. |
| 122 | client.on.on_message = [&](uint64_t, uint64_t, uint64_t, uint64_t, |
| 123 | const uint8_t *bytes, size_t len) { |
| 124 | legacy.broadcast(bytes, len); |
| 125 | to_local++; |
| 126 | }; |
| 127 | client.on.on_disconnect = [](const std::string &why) { |
| 128 | fprintf(stderr, "[swarmd-session] lost the daemon: %s\n", why.c_str()); |
| 129 | }; |
| 130 | |
| 131 | // A frame from a local client goes to the peers AND to the other locals — |
| 132 | // session-daemon fans out both ways (session_daemon.cpp:189) and a second |
| 133 | // local client on one machine would otherwise never hear the first. |
| 134 | legacy.on_frame = [&](uint64_t from, const uint8_t *data, size_t len) { |
| 135 | std::vector<uint8_t> copy(data, data + len); |
| 136 | legacy.broadcast(copy.data(), copy.size(), from); |
| 137 | // `accepted == false` means no peer has the channel open yet. That is |
| 138 | // normal before the swarm meshes and is not an error to report per frame. |
| 139 | if (client.send(channel, 0, 0, copy.data(), copy.size())) to_peers++; |
| 140 | }; |
| 141 | |
| 142 | printf("[swarmd-session] listening on %s (swarmd at %s)\n", sock, swarmd_sock.c_str()); |
| 143 | fflush(stdout); |
| 144 | |
| 145 | // ⛔ SAY WHETHER THE SWARM IS REACHABLE, not just whether we started. "Bound" |
| 146 | // is a local fact; two peers once sat on one topic for ten minutes both |
| 147 | // reporting success while neither had ever bootstrapped. |
| 148 | bool said_up = false; |
| 149 | int ticks = 0; |
| 150 | while (!g_stop) { |
| 151 | legacy.poll(5); |
| 152 | if (!client.poll(5)) break; |
| 153 | |
| 154 | if ((ticks++ % 200) == 0) { |
| 155 | swarmd::Status st; |
| 156 | if (client.status(st)) { |
| 157 | if (!said_up && st.bootstrapped) { |
| 158 | printf("[swarmd-session] dht=up\n"); |
| 159 | said_up = true; |
| 160 | } |
| 161 | printf("[swarmd-session] dht=%s peers=%llu local=%d " |
| 162 | "to_peers=%llu to_local=%llu\n", |
| 163 | st.bootstrapped ? "up" : "connecting", |
| 164 | (unsigned long long)st.peers_connected, legacy.client_count(), |
| 165 | (unsigned long long)to_peers, (unsigned long long)to_local); |
| 166 | fflush(stdout); |
| 167 | } |
| 168 | } |
| 169 | } |
| 170 | |
| 171 | printf("[swarmd-session] stopping\n"); |
| 172 | legacy.close_all(); |
| 173 | client.close(); |
| 174 | stop_daemon.store(true); |
| 175 | daemon_thread.join(); |
| 176 | daemon.stop(); |
| 177 | ::unlink(sock); |
| 178 | ::unlink(swarmd_sock.c_str()); |
| 179 | return 0; |
| 180 | } |