// Gate: a PeerConnection must outlive its udx stream's libuv handles.
//
// udx_stream_t embeds three libuv handles (timer, refill_pacing_timer,
// pending_packet_prepare). udx_stream_destroy closes them ASYNCHRONOUSLY and says so
// through the stream's on_finalize (libudx udx.c:355-361, 480-482). The transport used to
// `delete` the PeerConnection — which holds the udx_stream_t by value — right after
// udx_stream_destroy, on disconnect_peer and on connection dedup. libuv then walked its
// closing queue through freed memory: uv__finish_close's assertion
// (`handle->flags & UV_HANDLE_CLOSING`), seen by keetd's M3d run on the public DHT.
//
// Drives the deterministic path, disconnect_peer, three times on two meshed transports.
// Build with SAN=address to see the use-after-free directly; a plain build can pass by
// luck (freed memory not yet reused), which is why the ASan run is the gate.
//
// node ../integration/layer2_testnet.js &
// SAN=address ./build_tests.sh && ./test_peer_lifecycle [host:port]
#include "noise_ik.h"
#include "udx_transport.h"
#include <sodium.h>
#include <atomic>
#include <chrono>
#include <cstdio>
#include <cstring>
#include <fstream>
#include <mutex>
#include <string>
#include <thread>
#include <unistd.h>
static int passed = 0, failed = 0;
static void check (bool ok, const char *what) { (ok ? passed : failed)++; printf(" %s %s\n", ok ? "PASS" : "FAIL", what); }
static void sleep_ms (int ms) { std::this_thread::sleep_for(std::chrono::milliseconds(ms)); }
template <typename F> static bool wait_for (F f, int ms) {
for (int i = 0; i < ms / 10; i++) { if (f()) return true; sleep_ms(10); }
return f();
}
int main (int argc, char **argv) {
if (sodium_init() < 0) return 2;
std::string bs;
if (argc > 1) bs = argv[1];
else { std::ifstream f("/tmp/layer_bootstrap.txt"); if (f) std::getline(f, bs); }
if (bs.find(',') != std::string::npos) bs = bs.substr(0, bs.find(','));
while (!bs.empty() && isspace((unsigned char)bs.back())) bs.pop_back();
if (bs.empty()) { printf("FAIL: no bootstrap (node ../integration/layer2_testnet.js &)\n"); return 1; }
UdxTransport a, b;
std::mutex m;
int32_t aPeer = 0;
std::atomic<int> aReady{ 0 }, bReady{ 0 }, bGone{ 0 };
a.set_mux_callbacks([&] (int32_t id, Protomux *) { std::lock_guard<std::mutex> l(m); aPeer = id; aReady++; }, nullptr);
b.set_mux_callbacks([&] (int32_t, Protomux *) { bReady++; }, [&] (int32_t) { bGone++; });
uint8_t topic[32];
char seed[64];
snprintf(seed, sizeof(seed), "peer-lifecycle/%d", (int)getpid());
crypto_generichash(topic, 32, (const uint8_t *)seed, strlen(seed), nullptr, 0);
const char *bsv[1] = { bs.c_str() };
Ed25519KeyPair ka = NoiseIK::generate_keypair(), kb = NoiseIK::generate_keypair();
a.start_dht(ka, bsv, 1);
b.start_dht(kb, bsv, 1);
check(wait_for([&] { return a.is_dht_active() && b.is_dht_active(); }, 15000), "both DHTs active");
a.swarm_join(topic, true, true);
b.swarm_join(topic, true, true);
for (int round = 1; round <= 3; round++) {
const int want = round;
const bool up = wait_for([&] { return aReady >= want && bReady >= want; }, 40000);
char what[80];
snprintf(what, sizeof(what), "round %d: meshed", round);
check(up, what);
if (!up) break;
int32_t pid;
{ std::lock_guard<std::mutex> l(m); pid = aPeer; }
const int gone = bGone;
a.disconnect_peer(pid, true); // destroy + remove: the path that freed too early
snprintf(what, sizeof(what), "round %d: the remote sees the disconnect", round);
check(wait_for([&] { return bGone > gone; }, 30000), what);
sleep_ms(500); // let the loop run its closing handles
snprintf(what, sizeof(what), "round %d: both transports still running", round);
check(a.is_running() && b.is_running(), what);
}
a.stop();
b.stop();
printf("\n%s test_peer_lifecycle: %d/%d\n", failed ? "FAIL" : "PASS", passed, passed + failed);
return failed ? 1 : 0;
}