// The blender-bridge SESSION CHANNEL, on the real swarm, without Godot.
//
// blender-bridge/docs/P2P_BRIDGE_SPEC.md §5.5. This is the last unproven link
// between 85 pure assertions in blender-bridge/session/ and a session that could
// actually run: everything above this is decided logic, everything below is the
// transport, and this is the seam.
//
// ⛔ WHAT THIS PROVES AND WHAT IT DOES NOT.
//
// It proves the CHANNEL: two peers find each other on a topic, a Protomux channel
// named `blender-bridge-session/v1` opens on both, and an opaque frame arrives
// byte-identical at the other end. It proves NOTHING about grab arbitration --
// that lives in blender-bridge/session/grab.py, is gated there, and is
// deliberately not reimplemented in the transport. This layer never parses a
// frame; it does not know what a grab is.
//
// ⚠ The transport underneath is not on trial here. It is production-proven --
// two hyper-tps instances play over the public DHT, 314 unit tests, 10/10
// integration layers. If this file fails, suspect the CHANNEL wiring, not the DHT.
//
// Modelled on test_three_peer_swarm.cpp, which is the existing Godot-free way to
// drive UdxTransport/SwarmManager and attach a plain-C++ protocol to the mux
// (there, hyperoverlay; here, the session channel). set_mux_callbacks takes
// std::function, so no Godot is involved.
//
// Build/run:
// node ../integration/layer2_testnet.js & # writes /tmp/layer_bootstrap.txt
// ./build_tests.sh
// ./test_session_channel [host:port]
#include <chrono>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <fstream>
#include <map>
#include <mutex>
#include <string>
#include <thread>
#include <vector>
#include <unistd.h>
#include <sodium.h>
#include "noise_ik.h"
#include "protomux.h"
#include "udx_transport.h"
// NOTE: UdxTransport/Protomux are in the global namespace; only the overlay
// pieces live under `hyperswarm::`, and this file attaches no overlay.
// The protocol name carries its version, the way keet-call-presence/v3 does: a
// peer speaking another version simply never pairs, and no message has to carry
// or negotiate one.
static const char *SESSION_PROTOCOL = "blender-bridge-session/v1";
static int g_pass = 0;
static int g_fail = 0;
static void check(bool ok, const char *name) {
if (ok) {
g_pass++;
printf(" PASS: %s\n", name);
} else {
g_fail++;
printf(" FAIL: %s\n", name);
}
}
static void sleep_ms(int ms) {
std::this_thread::sleep_for(std::chrono::milliseconds(ms));
}
// One peer: transport plus a session channel per connected mux.
struct Node {
UdxTransport t;
std::string label;
// on_mux_ready / on_mux_gone fire on the UDX THREAD while the test thread
// reads these, so both need the lock.
std::mutex m;
std::map<int32_t, ProtomuxChannel *> channels;
int opened = 0;
std::vector<std::vector<uint8_t>> received;
std::vector<uint8_t> topic_id;
// Stop the transport BEFORE the members its callbacks use are destroyed. `t`
// is declared first, so it would otherwise be destroyed last: its thread's
// final stream teardown fired on_mux_gone into a map already freed (ASan:
// heap-use-after-free in the gone lambda, on every exit path).
~Node() { t.stop(); }
void wire(const uint8_t *topic) {
topic_id.assign(topic, topic + 32);
t.set_mux_callbacks(
[this](int32_t id, Protomux *mux) {
if (!mux) return;
std::lock_guard<std::mutex> lk(m);
if (channels.count(id)) return;
// {protocol, id} identifies the channel, so one mux can carry
// several sessions discriminated by topic.
ProtomuxChannel *ch = mux->create_channel(SESSION_PROTOCOL, topic_id);
if (ch == nullptr) return; // already open on this mux
// ONE message slot. The slot IS the message type, which is why
// the wire format carries a body rather than a type prefix on
// this hop. More types become more slots.
ch->add_message([this](const uint8_t *data, size_t len, ProtomuxChannel *) {
std::lock_guard<std::mutex> lk2(m);
received.push_back(std::vector<uint8_t>(data, data + len));
});
ch->on_open = [this](const uint8_t *, size_t) {
std::lock_guard<std::mutex> lk2(m);
opened++;
};
channels[id] = ch;
ch->open();
},
[this](int32_t id) {
// §5.4: a closing channel IS the disconnect. Whatever that peer
// held is released now, with no TTL to wait out.
std::lock_guard<std::mutex> lk(m);
channels.erase(id);
});
}
int open_channels() {
std::lock_guard<std::mutex> lk(m);
return opened;
}
size_t frames() {
std::lock_guard<std::mutex> lk(m);
return received.size();
}
bool send(const std::vector<uint8_t> &frame) {
std::lock_guard<std::mutex> lk(m);
bool any = false;
for (auto &kv : channels) {
if (kv.second->send(0, frame.data(), frame.size())) any = true;
}
return any;
}
};
static bool wait_dht(UdxTransport **ts, int n, int timeout_ms) {
int waited = 0;
while (waited < timeout_ms) {
bool all = true;
for (int i = 0; i < n; i++)
if (!ts[i]->is_dht_active()) all = false;
if (all) return true;
sleep_ms(50);
waited += 50;
}
return false;
}
static bool wait_connected(UdxTransport **ts, int n, int want, int timeout_ms) {
int waited = 0;
while (waited < timeout_ms) {
bool all = true;
for (int i = 0; i < n; i++)
if (ts[i]->get_swarm_status().peers_connected < want) all = false;
if (all) return true;
sleep_ms(50);
waited += 50;
}
return false;
}
static bool wait_channels(Node **ns, int n, int want, int timeout_ms) {
int waited = 0;
while (waited < timeout_ms) {
bool all = true;
for (int i = 0; i < n; i++)
if (ns[i]->open_channels() < want) all = false;
if (all) return true;
sleep_ms(50);
waited += 50;
}
return false;
}
int main(int argc, char **argv) {
if (sodium_init() < 0) {
printf("sodium_init failed\n");
return 1;
}
std::string bootstrap;
if (argc > 1) {
bootstrap = argv[1];
} else {
std::ifstream f("/tmp/layer_bootstrap.txt");
if (f) std::getline(f, bootstrap);
}
while (!bootstrap.empty() && (bootstrap.back() == '\n' || bootstrap.back() == '\r' ||
bootstrap.back() == ' '))
bootstrap.pop_back();
size_t comma = bootstrap.find(',');
if (comma != std::string::npos) bootstrap = bootstrap.substr(0, comma);
if (bootstrap.empty()) {
printf("FAIL: no bootstrap. Start the testnet first:\n");
printf(" node ../integration/layer2_testnet.js &\n");
return 1;
}
printf("=== Session channel over the swarm (bootstrap %s) ===\n", bootstrap.c_str());
// A topic unique to this run. Two runs sharing a topic would find each
// other's stale announces and the result would be an accident.
uint8_t topic[32];
char seed[80];
snprintf(seed, sizeof(seed), "blender-bridge-session/%d", (int)getpid());
crypto_generichash(topic, sizeof(topic), (const uint8_t *)seed, strlen(seed), nullptr, 0);
const char *bs[1] = { bootstrap.c_str() };
Node a, b;
a.label = "A";
b.label = "B";
a.wire(topic);
b.wire(topic);
Ed25519KeyPair ka = NoiseIK::generate_keypair();
Ed25519KeyPair kb = NoiseIK::generate_keypair();
a.t.start_dht(ka, bs, 1);
b.t.start_dht(kb, bs, 1);
UdxTransport *pair[2] = { &a.t, &b.t };
if (!wait_dht(pair, 2, 15000)) {
check(false, "both DHTs active");
printf("\n%d passed, %d failed\n", g_pass, g_fail);
return 1;
}
check(true, "both DHTs active");
a.t.swarm_join(topic, true, true);
b.t.swarm_join(topic, true, true);
bool meshed = wait_connected(pair, 2, 1, 30000);
check(meshed, "the two peers meshed on the topic");
if (!meshed) {
printf("\n%d passed, %d failed\n", g_pass, g_fail);
return 1;
}
Node *ns[2] = { &a, &b };
bool opened = wait_channels(ns, 2, 1, 15000);
check(opened, "the session channel opened on both peers");
if (!opened) {
printf(" A opened=%d B opened=%d\n", a.open_channels(), b.open_channels());
printf("\n%d passed, %d failed\n", g_pass, g_fail);
return 1;
}
// An opaque payload. The transport must not care what is in it: session
// frames are encoded by blender-bridge/session/wire.py, not here.
std::vector<uint8_t> frame = { 0x00, 0xaa, 0xbb, 0xcc, 0xdd, 0xee };
check(a.send(frame), "A sent a frame");
int waited = 0;
while (b.frames() == 0 && waited < 10000) {
sleep_ms(50);
waited += 50;
}
check(b.frames() == 1, "exactly one frame arrived at B");
bool identical = false;
{
std::lock_guard<std::mutex> lk(b.m);
identical = !b.received.empty() && b.received[0] == frame;
}
check(identical, "it arrived byte-identical");
// The reverse direction is a different code path on the responder side, and
// an untested direction is how bug #15 blocked every incoming connection to a
// C++ server while the client path passed.
std::vector<uint8_t> back = { 0x01, 0x11, 0x22 };
check(b.send(back), "B sent a frame back");
waited = 0;
while (a.frames() == 0 && waited < 10000) {
sleep_ms(50);
waited += 50;
}
bool back_ok = false;
{
std::lock_guard<std::mutex> lk(a.m);
back_ok = !a.received.empty() && a.received[0] == back;
}
check(back_ok, "the reverse direction carries a frame too");
printf("\n%d passed, %d failed\n", g_pass, g_fail);
return g_fail == 0 ? 0 : 1;
}