OverviewFilesCommitsIssuesDocsReleasesPeers

hyperswarm-peer commits / 8b811f5

udx: an opt-in stream consumer — the secret-stream without our Protomux
For an embedder that runs its own protocol stack over the connection (keetd: protomux 3.12.0 + hypercore replication, which must not sit on the 3.10.1 port). set_stream_consumer(open, frame, drain, close): a connection reaching noise_ready then gets NO internal Protomux and no godot-rpc channel; the consumer gets open(peer_id, is_initiator, handshake_hash, remote_public_key), each non-empty decrypted message (keepalives still eaten here), drain and close, on the UDX thread. send_frame / close_stream are its write side. Set before starting; open streams without a mux are replayed. Default path unchanged: with no consumer set nothing differs. Gate test_stream_consumer (DHT path, local testnet), 15/15: one open per side, opposite initiators, equal non-zero handshake hash, each other's key; frames both ways intact and ordered incl. 60 KiB; no internal mux; close both ends; send after close is refused; and without a consumer on_mux_ready still fires. Two mutations (consumer also gets the mux, frames not routed) red. Full suite green, session channel included. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
8b811f50caa2e0c0dff280b1af84fba036dce2d7
by plankatron <fleeky@gmail.com> · 2026-09-30T23:17:10.000Z

Changed files (4) +279 -0

src/udx_transport.cpp +68 -0
……@@ -452,6 +452,50 @@ void UdxTransport::_thread_func() {
452452 uv_loop_close(&_loop);
453453}
454454
455static UdxTransport::StreamInfo stream_info_of (const PeerConnection *peer) {
456 UdxTransport::StreamInfo info;
457 info.peer_id = peer->godot_peer_id;
458 info.is_initiator = peer->is_noise_initiator;
459 memcpy(info.handshake_hash, peer->_handshake_hash, 64);
460 memcpy(info.remote_public_key, peer->remote_public_key, 32);
461 return info;
462}
463
464void UdxTransport::set_stream_consumer(std::function<void(const StreamInfo &)> open,
465 std::function<void(int32_t, const uint8_t *, size_t)> frame,
466 std::function<void(int32_t)> drain,
467 std::function<void(int32_t)> close) {
468 run_on_loop([this, open = std::move(open), frame = std::move(frame), drain = std::move(drain),
469 close = std::move(close)]() mutable {
470 _stream_open = std::move(open);
471 _stream_frame = std::move(frame);
472 _stream_drain = std::move(drain);
473 _stream_close = std::move(close);
474 if (!_stream_open) return;
475 // CATCH UP, as set_mux_callbacks does: a stream that is already up and never
476 // got the internal mux becomes the consumer's.
477 for (auto &kv : _peers) {
478 PeerConnection *p = kv.second;
479 if (!p || !p->noise_ready || p->mux || p->stream_consumer) continue;
480 p->stream_consumer = true;
481 _stream_open(stream_info_of(p));
482 }
483 });
484}
485
486bool UdxTransport::send_frame(int32_t peer_id, const uint8_t *data, size_t len) {
487 PeerConnection *peer = _get_peer(peer_id);
488 if (!peer || !peer->connected || !peer->noise_ready || !peer->stream_consumer) return false;
489 _send_encrypted(peer, data, len);
490 return peer->writable;
491}
492
493void UdxTransport::close_stream(int32_t peer_id) {
494 PeerConnection *peer = _get_peer(peer_id);
495 if (!peer || !peer->connected) return;
496 udx_stream_destroy(&peer->stream); // _on_stream_close follows, which tells the consumer
497}
498
455499void UdxTransport::set_mux_callbacks(std::function<void(int32_t, Protomux *)> ready,
456500 std::function<void(int32_t)> gone) {
457501 run_on_loop([this, ready = std::move(ready), gone = std::move(gone)]() mutable {
……@@ -799,6 +843,12 @@ void UdxTransport::_dispatch_incoming(PeerConnection *peer,
799843 return;
800844 }
801845
846 // The stream consumer's connection: the message, as is.
847 if (peer->stream_consumer) {
848 if (_stream_frame) _stream_frame(peer->godot_peer_id, pt.data(), pt.size());
849 return;
850 }
851
802852 // Route through protomux if available.
803853 if (peer->mux) {
804854 DHT_LOG("[UdxTransport] dispatch: routing %zu decrypted bytes to protomux for peer %d\n",
……@@ -946,6 +996,8 @@ void UdxTransport::_on_stream_close(udx_stream_t *stream, int status) {
946996 transport->_peer_events.push_back({peer->godot_peer_id, false});
947997 if (transport->on_mux_gone) transport->on_mux_gone(peer->godot_peer_id);
948998 }
999 // Outside the events lock: a consumer may react by writing or closing.
1000 if (peer->stream_consumer && transport->_stream_close) transport->_stream_close(peer->godot_peer_id);
9491001
9501002 // Notify SwarmManager of disconnect for retry scheduling.
9511003 if (transport->_swarm_manager) {
……@@ -1001,6 +1053,10 @@ void UdxTransport::_on_stream_drain(udx_stream_t *stream) {
10011053 // Tell whoever parked. Protomux forwards it to its consumer — for the overlay
10021054 // that is the serve pump, which resumes from where it stopped.
10031055 if (peer->mux) peer->mux->notify_drain();
1056 if (peer->stream_consumer) {
1057 UdxTransport *transport = static_cast<UdxTransport *>(stream->socket->data);
1058 if (transport && transport->_stream_drain) transport->_stream_drain(peer->godot_peer_id);
1059 }
10041060}
10051061
10061062void UdxTransport::_on_write_ack(udx_stream_write_t *req, int status, int unordered) {
……@@ -1302,6 +1358,18 @@ void UdxTransport::_handle_noise_message(PeerConnection *peer, NoiseMessageType
13021358 DHT_LOG("[UdxTransport] noise_ready! peer=%d — encryption fully established\n",
13031359 peer->godot_peer_id);
13041360
1361 // A stream consumer takes the connection whole: no internal mux, no
1362 // godot-rpc. See set_stream_consumer.
1363 if (_stream_open) {
1364 peer->stream_consumer = true;
1365 {
1366 std::lock_guard<std::mutex> lock(_events_mutex);
1367 _peer_events.push_back({peer->godot_peer_id, true});
1368 }
1369 _stream_open(stream_info_of(peer));
1370 return;
1371 }
1372
13051373 // Create Protomux for this peer.
13061374 DHT_LOG("[UdxTransport] creating Protomux for peer %d\n", peer->godot_peer_id);
13071375 peer->mux = new Protomux();
src/udx_transport.h +40 -0
……@@ -104,6 +104,8 @@ struct PeerConnection {
104104 // fires. A late consumer is replayed against this rather than against `mux`
105105 // being non-null, so it is told exactly what a punctual one would have been.
106106 bool mux_ready = false;
107 // True when this connection belongs to the stream consumer (no internal mux).
108 bool stream_consumer = false;
107109
108110 // Receive buffer for length-prefixed framing (matches JS SecretStream's uint24 LE).
109111 // UDX reliable streams are byte streams — multiple writes can coalesce into one read.
……@@ -239,6 +241,44 @@ public:
239241 void set_mux_callbacks(std::function<void(int32_t, Protomux *)> ready,
240242 std::function<void(int32_t)> gone);
241243
244 // ---- Stream consumer: the connection WITHOUT our Protomux -----------------
245 //
246 // For an embedder that runs its own protocol stack over the secret-stream —
247 // keetd puts protomux 3.12.0 + hypercore replication there. OPT-IN: when no
248 // consumer is set, nothing below changes and every connection gets the
249 // internal Protomux + godot-rpc channel exactly as before.
250 //
251 // When a consumer IS set, a connection that reaches noise_ready gets NO
252 // internal Protomux and no godot-rpc channel. Instead, all on the UDX thread:
253 // open(info) once, at noise_ready
254 // frame(id, p, n) every non-empty decrypted secret-stream message
255 // (keepalives are still eaten here, as the reference does)
256 // drain(id) the stream fell back under its write watermark
257 // close(id) the stream closed
258 // Set it BEFORE starting: a connection that already got the internal mux keeps
259 // it. Streams already open without a mux are replayed to `open`.
260 struct StreamInfo {
261 int32_t peer_id;
262 bool is_initiator; // the Noise initiator — secret-stream's isInitiator
263 uint8_t handshake_hash[64]; // secret-stream's handshakeHash
264 uint8_t remote_public_key[32]; // secret-stream's remotePublicKey
265 };
266 void set_stream_consumer(std::function<void(const StreamInfo &)> open,
267 std::function<void(int32_t, const uint8_t *, size_t)> frame,
268 std::function<void(int32_t)> drain,
269 std::function<void(int32_t)> close);
270 // UDX THREAD ONLY (from a consumer callback, or run_on_loop). Encrypts and
271 // writes one message; returns false once the stream is past its watermark
272 // (wait for drain) or the peer is gone.
273 bool send_frame(int32_t peer_id, const uint8_t *data, size_t len);
274 // UDX THREAD ONLY. Destroys the stream; `close` follows.
275 void close_stream(int32_t peer_id);
276
277 std::function<void(const StreamInfo &)> _stream_open;
278 std::function<void(int32_t, const uint8_t *, size_t)> _stream_frame;
279 std::function<void(int32_t)> _stream_drain;
280 std::function<void(int32_t)> _stream_close;
281
242282 // Internal, but declared here rather than after a `private:` — the insertion
243283 // point sits inside a public section and closing it would silently privatise
244284 // every declaration below, which is exactly what happened the first time.
test/build_tests.sh +12 -0
……@@ -106,6 +106,18 @@ g++ -o test_regression \
106106 $UDX_OBJS $LDFLAGS
107107echo " OK: test_regression"
108108
109# ---- test_stream_consumer: the opt-in seam an embedder (keetd) uses to run its own
110# protocol stack over the secret-stream, and proof the default path is unchanged ----
111echo "Building test_stream_consumer..."
112g++ $CXXFLAGS -o $OBJ/test_stream_consumer.o test_stream_consumer.cpp
113g++ -o test_stream_consumer \
114 $OBJ/test_stream_consumer.o $OBJ/noise_ik.o $OBJ/noise_handshake.o \
115 $OBJ/noise_stream.o $OBJ/dht_rpc.o $OBJ/holepuncher.o \
116 $OBJ/hyperdht.o $OBJ/dht_connector.o $OBJ/udx_transport.o \
117 $OBJ/swarm_manager.o $OBJ/protomux.o \
118 $UDX_OBJS $LDFLAGS
119echo " OK: test_stream_consumer"
120
109121# ---- test_interop_client ----
110122echo "Building test_interop_client..."
111123g++ $CXXFLAGS -o $OBJ/test_interop_client.o test_interop_client.cpp
test/test_stream_consumer.cpp +159 -0
……@@ -0,0 +1,159 @@
1// Gate for UdxTransport's stream consumer (set_stream_consumer / send_frame /
2// close_stream): the seam keetd uses to run its own protomux over the secret-stream.
3//
4// Over the DHT path (the production one), against a local JS testnet:
5// node ../integration/layer2_testnet.js & # writes /tmp/layer_bootstrap.txt
6// ./test_stream_consumer [host:port]
7// Two transports on one topic, BOTH with a consumer:
8// - each sees exactly one open; the initiator flags are opposite, the handshake
9// hashes equal, and each side's remote key is the other's
10// - frames arrive intact and in order both ways, including one of 60 KiB
11// - NO internal Protomux: on_mux_ready never fires
12// - close_stream on one side closes the stream, and `close` fires
13// And a pair WITHOUT a consumer: the default path is unchanged — on_mux_ready fires
14// (the godot-rpc channel opened), and no consumer callback does.
15#include "noise_ik.h"
16#include "udx_transport.h"
17
18#include <sodium.h>
19
20#include <atomic>
21#include <chrono>
22#include <cstdio>
23#include <cstring>
24#include <fstream>
25#include <string>
26#include <unistd.h>
27#include <mutex>
28#include <thread>
29#include <vector>
30
31static int passed = 0, failed = 0;
32static void check (bool ok, const char *what) {
33 (ok ? passed : failed)++;
34 printf(" %s %s\n", ok ? "PASS" : "FAIL", what);
35}
36static void sleep_ms (int ms) { std::this_thread::sleep_for(std::chrono::milliseconds(ms)); }
37template <typename F> static bool wait_for (F f, int ms) {
38 for (int i = 0; i < ms / 10; i++) { if (f()) return true; sleep_ms(10); }
39 return f();
40}
41
42struct Side {
43 std::mutex m;
44 std::vector<UdxTransport::StreamInfo> opens;
45 std::vector<std::vector<uint8_t>> frames;
46 std::atomic<int> closes{ 0 }, muxReady{ 0 };
47 void attach (UdxTransport &t) {
48 t.set_stream_consumer(
49 [this] (const UdxTransport::StreamInfo &i) { std::lock_guard<std::mutex> l(m); opens.push_back(i); },
50 [this] (int32_t, const uint8_t *p, size_t n) { std::lock_guard<std::mutex> l(m); frames.emplace_back(p, p + n); },
51 [] (int32_t) {},
52 [this] (int32_t) { closes++; });
53 t.set_mux_callbacks([this] (int32_t, Protomux *) { muxReady++; }, nullptr);
54 }
55 size_t nopens () { std::lock_guard<std::mutex> l(m); return opens.size(); }
56 size_t nframes () { std::lock_guard<std::mutex> l(m); return frames.size(); }
57};
58
59static std::vector<uint8_t> pattern (size_t n, uint8_t seed) {
60 std::vector<uint8_t> v(n);
61 for (size_t i = 0; i < n; i++) v[i] = uint8_t(seed + i * 31);
62 return v;
63}
64
65static std::string g_bootstrap;
66
67// two transports on the testnet, joined to one fresh topic
68static bool mesh (UdxTransport &a, UdxTransport &b, Ed25519KeyPair &ka, Ed25519KeyPair &kb, const char *label) {
69 uint8_t topic[32];
70 char seed[80];
71 snprintf(seed, sizeof(seed), "stream-consumer/%s/%d", label, (int)getpid());
72 crypto_generichash(topic, 32, (const uint8_t *)seed, strlen(seed), nullptr, 0);
73 const char *bs[1] = { g_bootstrap.c_str() };
74 ka = NoiseIK::generate_keypair();
75 kb = NoiseIK::generate_keypair();
76 a.start_dht(ka, bs, 1);
77 b.start_dht(kb, bs, 1);
78 if (!wait_for([&] { return a.is_dht_active() && b.is_dht_active(); }, 15000)) return false;
79 a.swarm_join(topic, true, true);
80 b.swarm_join(topic, true, true);
81 return true;
82}
83
84static void test_consumer_pair () {
85 printf("\n--- stream consumer: two transports over the DHT ---\n");
86 UdxTransport a, b;
87 Side s, c;
88 s.attach(a);
89 c.attach(b);
90 Ed25519KeyPair ka, kb;
91 check(mesh(a, b, ka, kb, "consumer"), "both DHTs active, topic joined");
92
93 const bool opened = wait_for([&] { return s.nopens() == 1 && c.nopens() == 1; }, 30000);
94 check(opened, "each side saw exactly one open");
95 if (!opened) { printf(" opens: A=%zu B=%zu\n", s.nopens(), c.nopens()); a.stop(); b.stop(); return; }
96
97 UdxTransport::StreamInfo si, ci;
98 { std::lock_guard<std::mutex> l(s.m); si = s.opens[0]; }
99 { std::lock_guard<std::mutex> l(c.m); ci = c.opens[0]; }
100 check(si.is_initiator != ci.is_initiator, "initiator flags are opposite");
101 check(memcmp(si.handshake_hash, ci.handshake_hash, 64) == 0, "handshake hashes are equal");
102 bool nonzero = false;
103 for (int i = 0; i < 64; i++) nonzero |= si.handshake_hash[i] != 0;
104 check(nonzero, "handshake hash is not zeros");
105 check(memcmp(si.remote_public_key, kb.public_key, 32) == 0 && memcmp(ci.remote_public_key, ka.public_key, 32) == 0,
106 "each side's remote key is the other's");
107
108 // frames both ways, in order, one of 60 KiB
109 const std::vector<std::vector<uint8_t>> toA = { pattern(1, 1), pattern(300, 2), pattern(60 * 1024, 3), pattern(7, 4) };
110 const std::vector<std::vector<uint8_t>> toB = { pattern(5, 9), pattern(1000, 8) };
111 b.run_on_loop([&] { for (auto &f : toA) b.send_frame(ci.peer_id, f.data(), f.size()); });
112 a.run_on_loop([&] { for (auto &f : toB) a.send_frame(si.peer_id, f.data(), f.size()); });
113 const bool got = wait_for([&] { return s.nframes() == toA.size() && c.nframes() == toB.size(); }, 10000);
114 check(got, "every frame arrived");
115 { std::lock_guard<std::mutex> l(s.m); check(s.frames == toA, "A received B's frames intact and in order"); }
116 { std::lock_guard<std::mutex> l(c.m); check(c.frames == toB, "B received A's frames intact and in order"); }
117
118 check(s.muxReady == 0 && c.muxReady == 0, "no internal Protomux (on_mux_ready never fired)");
119
120 // close from B: both see the stream close
121 b.run_on_loop([&] { b.close_stream(ci.peer_id); });
122 check(wait_for([&] { return c.closes >= 1; }, 5000), "close_stream fires close locally");
123 check(wait_for([&] { return s.closes >= 1; }, 30000), "the remote sees the stream close");
124
125 // send_frame to a peer that is gone answers false
126 std::atomic<int> after{ -1 };
127 b.run_on_loop([&] { uint8_t x = 1; after = b.send_frame(ci.peer_id, &x, 1) ? 1 : 0; });
128 check(wait_for([&] { return after != -1; }, 2000) && after == 0, "send_frame to a closed peer answers false");
129
130 a.stop();
131 b.stop();
132}
133
134static void test_default_path_unchanged () {
135 printf("\n--- no consumer: the default path is unchanged ---\n");
136 UdxTransport a, b;
137 std::atomic<int> aReady{ 0 }, bReady{ 0 };
138 a.set_mux_callbacks([&] (int32_t, Protomux *) { aReady++; }, nullptr);
139 b.set_mux_callbacks([&] (int32_t, Protomux *) { bReady++; }, nullptr);
140 Ed25519KeyPair ka, kb;
141 check(mesh(a, b, ka, kb, "default"), "both DHTs active, topic joined");
142 check(wait_for([&] { return aReady >= 1 && bReady >= 1; }, 30000), "on_mux_ready fired on both sides (godot-rpc opened)");
143 a.stop();
144 b.stop();
145}
146
147int main (int argc, char **argv) {
148 if (sodium_init() < 0) return 2;
149 if (argc > 1) g_bootstrap = argv[1];
150 else { std::ifstream f("/tmp/layer_bootstrap.txt"); if (f) std::getline(f, g_bootstrap); }
151 const size_t comma = g_bootstrap.find(',');
152 if (comma != std::string::npos) g_bootstrap = g_bootstrap.substr(0, comma);
153 while (!g_bootstrap.empty() && isspace((unsigned char)g_bootstrap.back())) g_bootstrap.pop_back();
154 if (g_bootstrap.empty()) { printf("FAIL: no bootstrap. Start the testnet: node ../integration/layer2_testnet.js &\n"); return 1; }
155 test_consumer_pair();
156 test_default_path_unchanged();
157 printf("\n%s test_stream_consumer: %d/%d\n", failed ? "FAIL" : "PASS", passed, passed + failed);
158 return failed ? 1 : 0;
159}