code wiki / _hdl_build / nx_signaling_v2.nx
nx_signaling_v2.nx source
↩ module page · 1070 lines · 53459 B
1// nx_signaling_v2.nx -- bits-up N-party WebSocket signaling daemon for /video.
2//
3// Generation 2 of bench/nx_signaling.nx (1:1 rooms). What changed and why:
4// - N-party rooms (NX_SIG2_MAX_PEERS=8 fds per room) -- the family call is
5// more than two people. Relay = broadcast to every OTHER peer in the
6// room; the browser JS addresses peers by its own from/to envelope
7// fields and ignores frames not meant for it (server stays JSON-free,
8// single responsibility: room membership + relay only).
9// - Binds 127.0.0.1 ONLY. TLS termination happens in the sites daemon's
10// fork-per-connection child, which pumps decrypted bytes to this
11// loopback daemon (wss://nishifamily.com/signal/<room> on :8443) -- so
12// the plaintext WS port is never exposed off-host.
13// - Composes the SAME KAT'd primitives as v1: nx_websocket_upgrade
14// (RFC 6455 4.1), nx_websocket_stream (5.3), nx_websocket_frame (5.7),
15// nx_poll.
16//
17// Deploy: nx_sov_build_run nx_signaling_v2 -> push .sov.elf to the NAS.
18// Gate: nx_signaling_v2_gate.nx (3-peer mesh relay proof over loopback).
19// license_tier: ORIGINAL
20
21import "nx_syscalls.nx"
22import "nx_signal.nx" // SIGPIPE-ignore: a write to a dead peer socket must be an EPIPE error, never death
23import "nx_dgram_media.nx" // R2 (task #15): the SOVEREIGN UDP plane -- fragmentation/reassembly/dedupe wire
24import "nx_sfu_select.nx" // SFU (task #41): selective forwarding -- per-(sender,receiver) layer + seq-rewrite
25import "nx_room_key.nx" // room capability tokens: sha256(base_room||secret) links; data-driven protected list
26import "nx_websocket_frame.nx"
27import "nx_websocket_stream.nx"
28import "nx_websocket_upgrade.nx"
29import "nx_poll.nx"
30import "nx_sig2_txq.nx" // frame-atomic non-blocking sends (kills head-of-line stall + torn-frame poison)
31
32const NX_SIG2_PORT: i64 = 8445
33const NX_SIG2_MAX_ROOMS: i64 = 64
34const NX_SIG2_MAX_PEERS: i64 = 8
35const NX_SIG2_MAX_CONNS: i64 = 128
36const NX_SIG2_HDR_BUF_BYTES: i64 = 4096
37// 64KB: the relay IS the media plane (pure-Nishi video = JPEG frames +
38// PCM chunks as binary WS frames; no WebRTC). A 320x240 JPEG is 6-15KB,
39// quality bumps and roster JSON stay well inside 64KB.
40const NX_SIG2_FRAME_BUF_BYTES: i64 = 65536
41const NX_SIG2_ROOM_ID_POOL_BYTES: i64 = 4096
42const NX_SIG2_ACCEPT_BACKLOG: i64 = 32
43
44// Per-room state: up to MAX_PEERS fds. used=0 -> free slot.
45// Fixed layout: id_off, id_len, used, npeers, fd[8] = 12 i64 = 96 bytes.
46const NX_SIG2_ROOM_BYTES: i64 = 96
47
48func sig2_room_ptr(rooms: *u8, idx: i64) -> *i64 {
49 return (rooms as i64 + idx * NX_SIG2_ROOM_BYTES) as *i64
50}
51// field offsets (in i64 units)
52const R_ID_OFF: i64 = 0
53const R_ID_LEN: i64 = 1
54const R_USED: i64 = 2
55const R_NPEERS: i64 = 3
56const R_FD0: i64 = 4 // fd[i] at R_FD0 + i
57
58// Per-conn state: fd, room_idx.
59const NX_SIG2_CONN_BYTES: i64 = 16
60
61func sig2_conn_ptr(conns: *u8, idx: i64) -> *i64 {
62 return (conns as i64 + idx * NX_SIG2_CONN_BYTES) as *i64
63}
64
65// sockaddr_in for 127.0.0.1:<port> (loopback ONLY -- see header comment).
66func sig2_sockaddr_loopback(addr: *u8, port: i64) -> i64 {
67 addr[0] = 2 as u8; addr[1] = 0 as u8
68 addr[2] = ((port >> 8) & 0xff) as u8
69 addr[3] = (port & 0xff) as u8
70 addr[4] = 127 as u8; addr[5] = 0 as u8; addr[6] = 0 as u8; addr[7] = 1 as u8
71 addr[8] = 0 as u8; addr[9] = 0 as u8; addr[10] = 0 as u8; addr[11] = 0 as u8
72 addr[12] = 0 as u8; addr[13] = 0 as u8; addr[14] = 0 as u8; addr[15] = 0 as u8
73 return 16
74}
75
76func sig2_find_room(rooms: *u8, id_pool: *u8, id: *u8, id_len: i64) -> i64 {
77 var i: i64 = 0
78 while i < NX_SIG2_MAX_ROOMS {
79 let r: *i64 = sig2_room_ptr(rooms, i)
80 if r[R_USED] == 1 {
81 if r[R_ID_LEN] == id_len {
82 var j: i64 = 0
83 var ok: i64 = 1
84 while j < id_len {
85 if id_pool[r[R_ID_OFF] + j] != id[j] { ok = 0; j = id_len } else { j = j + 1 }
86 }
87 if ok == 1 { return i }
88 }
89 }
90 i = i + 1
91 }
92 return 0 - 1
93}
94
95func sig2_alloc_room(rooms: *u8, id_pool: *u8, id_pool_off_p: *i64, id: *u8, id_len: i64) -> i64 {
96 var i: i64 = 0
97 while i < NX_SIG2_MAX_ROOMS {
98 let r: *i64 = sig2_room_ptr(rooms, i)
99 if r[R_USED] == 0 {
100 let off: i64 = id_pool_off_p[0]
101 if off + id_len > NX_SIG2_ROOM_ID_POOL_BYTES { return 0 - 1 }
102 var k: i64 = 0
103 while k < id_len { id_pool[off + k] = id[k]; k = k + 1 }
104 id_pool_off_p[0] = off + id_len
105 r[R_ID_OFF] = off
106 r[R_ID_LEN] = id_len
107 r[R_USED] = 1
108 r[R_NPEERS] = 0
109 var p: i64 = 0
110 while p < NX_SIG2_MAX_PEERS { r[R_FD0 + p] = 0 - 1; p = p + 1 }
111 return i
112 }
113 i = i + 1
114 }
115 return 0 - 1
116}
117
118// Join: first free peer slot. Returns slot or -1 (room full).
119func sig2_join(rooms: *u8, ridx: i64, fd: i64) -> i64 {
120 let r: *i64 = sig2_room_ptr(rooms, ridx)
121 var p: i64 = 0
122 while p < NX_SIG2_MAX_PEERS {
123 if r[R_FD0 + p] == 0 - 1 {
124 r[R_FD0 + p] = fd
125 r[R_NPEERS] = r[R_NPEERS] + 1
126 return p
127 }
128 p = p + 1
129 }
130 return 0 - 1
131}
132
133// Leave: clear fd's slot; free the room when empty.
134func sig2_leave(rooms: *u8, ridx: i64, fd: i64) -> i64 {
135 let r: *i64 = sig2_room_ptr(rooms, ridx)
136 var p: i64 = 0
137 while p < NX_SIG2_MAX_PEERS {
138 if r[R_FD0 + p] == fd {
139 r[R_FD0 + p] = 0 - 1
140 r[R_NPEERS] = r[R_NPEERS] - 1
141 }
142 p = p + 1
143 }
144 if r[R_NPEERS] <= 0 { r[R_USED] = 0; r[R_NPEERS] = 0 }
145 return 0
146}
147
148// --- per-participant relay telemetry (the "where does it suck" SENSE layer) ---
149// tel[] = per (room,peer): [frames_relayed, send_fails]. A peer with rising send_fails
150// is the slow/wedged reader = the bottleneck participant (room-perf-arc: "relay wedges
151// when a peer's buffer fills"). Fire-and-forget (Cardinal 14): a logging failure must
152// NEVER disturb the relay. Emitted every NX_SIG2_EMIT_US by the main loop.
153const NX_SIG2_TEL_LOG: *u8 = "/volume1/homes/elderwesto/nishihost/room_telemetry.log"
154const NX_SIG2_EMIT_US: i64 = 5000000
155const NX_SIG2_TEL_SLOTS: i64 = 1024 // MAX_ROOMS(64) * MAX_PEERS(8) * 2
156
157// ---- R2 SOVEREIGN UDP PLANE (task #15; operator: pure nishi, hardware rung up, NO WebTransport) ----
158// Native clients (#22) + NishiOS speak nx_dgram_media datagrams straight to :8471; browser tabs stay on
159// the WSS lane (their sandbox has no raw UDP). SAME room registry, MIXED-PLANE rooms: the relay BRIDGES --
160// UDP->WSS: reassemble a dgram frame -> emit as one WS frame to every WSS member; WSS->UDP: fragment each
161// WS frame -> datagrams to every UDP member. The dgram payload IS the room-protocol frame (protocol SSOT
162// #21 unchanged); the dgram header is transport-only. JOIN: kind 0x4A datagram, payload = room id; the
163// client re-JOINs every ~5s (presence + NAT keepalive); members idle >30s are swept.
164const NX_SIG2_UDP_PORT: i64 = 8471
165const NX_SIG2_UDPM: i64 = 16 // UDP members (addr16 | room:i64 | last_us:i64 = 32B each)
166const SIG2_DG_JOIN: i64 = 0x4A
167const SIG2_DG_ACK: i64 = 0x4B
168const SIG2_DG_ROOMFRAME: i64 = 0x4D // payload = a room-protocol BINARY frame (bridged <-> WS binary)
169const SIG2_DG_TEXT: i64 = 0x54 // payload = a room-protocol TEXT frame (bridged <-> WS text JSON)
170const SIG2_DG_CASCADE: i64 = 0x4E // CASCADE-JOIN (relay-to-relay, Octo-class): enroll the sender as a
171// CASCADE member -> a frame from a cascade member fans to LOCAL peers
172// ONLY (split-horizon) so it crosses each inter-relay link exactly ONCE.
173
174func sig2_atoi(s: *u8) -> i64 {
175 var v: i64 = 0; var i: i64 = 0
176 while s[i] != (0 as u8) { let c: i64 = s[i] & 0xff; if c >= 48 { if c <= 57 { v = v * 10 + (c - 48) } } i = i + 1 }
177 return v }
178// parse dotted-decimal "a.b.c.d" -> out[0..3]
179func sig2_parse_ip(s: *u8, out: *u8) -> i64 {
180 var oi: i64 = 0; var v: i64 = 0; var i: i64 = 0; var run: i64 = 1
181 while run == 1 {
182 let c: i64 = s[i] & 0xff
183 if c == 0 { if oi < 4 { out[oi] = v as u8 } run = 0 }
184 else { if c == 46 { if oi < 4 { out[oi] = v as u8 } oi = oi + 1; v = 0 }
185 else { if c >= 48 { if c <= 57 { v = v * 10 + (c - 48) } } } }
186 i = i + 1
187 if i > 24 { run = 0 }
188 }
189 return 0 }
190
191func sig2_udpm_find(m: *u8, addr: *u8) -> i64 {
192 var i: i64 = 0
193 while i < NX_SIG2_UDPM {
194 let b: *u8 = ((m as i64) + i * 32) as *u8
195 let rp: *i64 = ((m as i64) + i * 32 + 16) as *i64
196 if rp[0] >= 0 {
197 var eq: i64 = 1
198 var k: i64 = 0
199 while k < 8 { if b[k] != addr[k] { eq = 0; k = 8 } else { k = k + 1 } }
200 if eq == 1 { return i }
201 }
202 i = i + 1
203 }
204 return 0 - 1 }
205func sig2_udpm_upsert(m: *u8, addr: *u8, ridx: i64, now: i64) -> i64 {
206 var i: i64 = sig2_udpm_find(m, addr)
207 if i < 0 {
208 var j: i64 = 0
209 while j < NX_SIG2_UDPM {
210 if i < 0 {
211 let rp: *i64 = ((m as i64) + j * 32 + 16) as *i64
212 if rp[0] < 0 { i = j }
213 if rp[0] >= 0 { if now - rp[1] > 30000000 { i = j } } // sweep stale >30s
214 }
215 j = j + 1
216 }
217 }
218 if i < 0 { return 0 - 1 }
219 let b: *u8 = ((m as i64) + i * 32) as *u8
220 var k: i64 = 0
221 while k < 16 { b[k] = addr[k]; k = k + 1 }
222 let rp: *i64 = ((m as i64) + i * 32 + 16) as *i64
223 rp[0] = ridx
224 rp[1] = now
225 return i }
226// fragment+send one room-frame to every UDP member of ridx (except member index `except`).
227func sig2_udp_fanout(ufd: i64, m: *u8, ridx: i64, except: i64, dkind: i64, body: *u8, blen: i64,
228 packbuf: *u8, relay_seq_p: *i64) -> i64 {
229 let np: i64 = dgm_pack_frame(dkind, "RELAY000" as *u8, relay_seq_p[0], body, blen, packbuf, 131072)
230 if np <= 0 { return 0 }
231 relay_seq_p[0] = relay_seq_p[0] + 1
232 var i: i64 = 0
233 while i < NX_SIG2_UDPM {
234 let rp: *i64 = ((m as i64) + i * 32 + 16) as *i64
235 if rp[0] == ridx { if i != except {
236 let dst: *u8 = ((m as i64) + i * 32) as *u8
237 var o: i64 = 0
238 var p: i64 = 0
239 while p < np {
240 let pl: i64 = (packbuf[o] & 0xff) + ((packbuf[o+1] & 0xff) << 8)
241 sys_sendto(ufd, ((packbuf as i64) + o + 2) as *u8, pl, 0, dst, 16)
242 o = o + 2 + pl
243 p = p + 1
244 }
245 } }
246 i = i + 1
247 }
248 return 0 }
249
250func sig2_telw(fd: i64, s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } sys_write(fd, s, n); return 0 }
251func sig2_teln(fd: i64, v: i64) -> i64 { let bb: *u8 = sys_mmap(28); var m: i64=v; if m<0 {m=0-m} let t: *u8 = sys_mmap(28); var k: i64=0; if m==0 {t[0]=48;k=1} while m>0 {t[k]=(48+(m%10)) as u8; m=m/10; k=k+1} var i: i64=0; while i<k {bb[i]=t[k-1-i]; i=i+1} sys_write(fd, bb, k); return 0 }
252
253// Emit one telemetry line per ACTIVE room: peers + per-peer relayed/fails. Cumulative
254// (rising fails on a peer localizes the bottleneck). Guarded; never crashes the relay.
255func sig2_emit_tel(rooms: *u8, tel: *i64, now: i64) -> i64 {
256 let lf: i64 = sys_openat_append(NX_SIG2_TEL_LOG, 420)
257 if lf < 0 { return 0 }
258 var ri: i64 = 0
259 while ri < NX_SIG2_MAX_ROOMS {
260 let r: *i64 = sig2_room_ptr(rooms, ri)
261 if r[R_USED] == 1 {
262 sig2_telw(lf, "ROOMTEL t=" as *u8); sig2_teln(lf, now)
263 sig2_telw(lf, " room=" as *u8); sig2_teln(lf, ri)
264 sig2_telw(lf, " peers=" as *u8); sig2_teln(lf, r[R_NPEERS])
265 var p: i64 = 0
266 while p < NX_SIG2_MAX_PEERS {
267 if r[R_FD0 + p] != 0 - 1 {
268 let ti: i64 = (ri * NX_SIG2_MAX_PEERS + p) * 2
269 sig2_telw(lf, " p" as *u8); sig2_teln(lf, p)
270 sig2_telw(lf, ":relayed=" as *u8); sig2_teln(lf, tel[ti])
271 sig2_telw(lf, ",fails=" as *u8); sig2_teln(lf, tel[ti + 1])
272 }
273 p = p + 1
274 }
275 sig2_telw(lf, "\n" as *u8)
276 }
277 ri = ri + 1
278 }
279 sys_close(lf)
280 return 0
281}
282
283// conn slot of an fd. -1 = not a WSS conn (e.g. UDP-bridge origin).
284func sig2_slot_of_fd(conns: *u8, fd: i64) -> i64 {
285 if fd < 0 { return 0 - 1 }
286 var i: i64 = 0
287 while i < NX_SIG2_MAX_CONNS {
288 let c: *i64 = sig2_conn_ptr(conns, i)
289 if c[0] == fd { return i }
290 i = i + 1
291 }
292 return 0 - 1 }
293
294// SFU pairs are keyed by MEMBER id (the 8B id every binary frame carries), NOT conn slot: the TIER-2
295// client hops sticky lanes (separate conns, switch at a key) and reconnects -- the pair's out_seq must
296// stay MONOTONIC across those or the receiver's dedupe would drop everything after a lane switch.
297// find-or-claim; a wrap-claim recycles the oldest slot and clears its pairs (organ reset = safe reuse).
298func sig2_memidx(sfu_mem: *i64, mem_ids: *u8, mcount_p: *i64, id: *u8) -> i64 {
299 var i: i64 = 0
300 var n: i64 = mcount_p[0]
301 if n > SFU_MAXC { n = SFU_MAXC }
302 while i < n {
303 let e: *u8 = ((mem_ids as i64) + i * 8) as *u8
304 var eq: i64 = 1
305 var q: i64 = 0
306 while q < 8 { if e[q] != id[q] { eq = 0; q = 8 } else { q = q + 1 } }
307 if eq == 1 { return i }
308 i = i + 1
309 }
310 let idx: i64 = mcount_p[0] % SFU_MAXC
311 mcount_p[0] = mcount_p[0] + 1
312 if mcount_p[0] > SFU_MAXC { sfu_reset_conn(sfu_mem, idx) } // recycled slot -> stale pairs cleared
313 let d: *u8 = ((mem_ids as i64) + idx * 8) as *u8
314 var w: i64 = 0
315 while w < 8 { d[w] = id[w]; w = w + 1 }
316 return idx }
317
318// Broadcast a data frame to every OTHER peer in the room; counts per-peer relayed/fails into tel[].
319// SFU (task #41): 0x57 video frames get a per-receiver verdict (nx_sfu_select: which LAYER this receiver
320// takes, switch only at target-layer key) + a rewritten per-pair seq so the forwarded subset stays
321// dedupe/P-chain-legal. Text/audio/chat/FEC broadcast as before; UDP-bridged frames (from_fd=-1 -> no
322// sender slot) and default-HI pairs behave EXACTLY like the old relay by construction.
323// room candidate list for activity ranking: member idxs of every conn in room ridx (deduped by memidx).
324// Returns count. cand cap = NX_SIG2_MAX_CONNS.
325func sig2_room_cand(conns: *u8, cmx: *i64, ridx: i64, cand: *i64) -> i64 {
326 var n: i64 = 0
327 var i: i64 = 0
328 while i < NX_SIG2_MAX_CONNS {
329 let c: *i64 = sig2_conn_ptr(conns, i)
330 if c[0] >= 0 { if c[1] == ridx { if cmx[i] >= 0 {
331 var dup: i64 = 0
332 var q: i64 = 0
333 while q < n { if cand[q] == cmx[i] { dup = 1; q = n } else { q = q + 1 } }
334 if dup == 0 { cand[n] = cmx[i]; n = n + 1 }
335 } } }
336 i = i + 1
337 }
338 return n }
339
340// sfuctx: [0]=sfu_mem [1]=conns [2]=conn_ids [3]=mem_ids [4]=mcount_p [5]=conn_memidx(*i64)
341// [6]=dom state (*i64: per room [dom_midx, last_check_us] x NX_SIG2_MAX_ROOMS) [7]=cand scratch.
342// Sender identity = the frame's OWN id bytes [1..8] (works for any origin, lane-stable). Receiver identity
343// = the memidx learned for that conn; a receiver with no learned id yet gets the frame PLAIN (it cannot
344// have subscribed, and plain = default-HI behavior with the sender's original seq -- keys resync later).
345func sig2_broadcast(rooms: *u8, ridx: i64, from_fd: i64,
346 fin: i64, opcode: i64, payload: *u8, payload_len: i64, tel: *i64, sfuctx: *i64) -> i64 {
347 let r: *i64 = sig2_room_ptr(rooms, ridx)
348 var is_vid: i64 = 0
349 var vlayer: i64 = 0
350 var vkey: i64 = 0
351 var smidx: i64 = 0 - 1
352 if opcode == 2 {
353 let vinf: *i64 = (sfuctx[9]) as *i64
354 if sfu_frame_bits(payload, payload_len, vinf) == 0 {
355 smidx = sig2_memidx((sfuctx[0]) as *i64, (sfuctx[3]) as *u8, (sfuctx[4]) as *i64,
356 ((payload as i64) + 1) as *u8)
357 if smidx >= 0 { is_vid = 1; vlayer = vinf[0]; vkey = vinf[1]
358 sfu_video_seen((sfuctx[0]) as *i64, smidx, sys_now_us()) // rank tie-break freshness
359 sig2_lprod_eval(sfuctx, smidx) } // R5 self-correcting: leavers/joins re-verdict on the sender's own frames
360 }
361 }
362 let cmx: *i64 = (sfuctx[5]) as *i64
363 let cand: *i64 = (sfuctx[7]) as *i64
364 var ncand: i64 = 0
365 if is_vid == 1 { ncand = sig2_room_cand((sfuctx[1]) as *u8, cmx, ridx, cand) }
366 var p: i64 = 0
367 while p < NX_SIG2_MAX_PEERS {
368 let pfd: i64 = r[R_FD0 + p]
369 if pfd != 0 - 1 {
370 if pfd != from_fd {
371 var send: i64 = 1
372 let rcs: i64 = sig2_slot_of_fd((sfuctx[1]) as *u8, pfd)
373 if is_vid == 1 {
374 if rcs >= 0 {
375 let rmidx: i64 = cmx[rcs]
376 if rmidx >= 0 {
377 // LAST-N first (R4): outside the receiver's top-N active senders -> not sent,
378 // and the pair's layer/seq state stays untouched (re-entry resyncs at a key).
379 if sfu_video_rank_ok((sfuctx[0]) as *i64, smidx, rmidx, cand, ncand, sys_now_us()) == 0 { send = 0 }
380 else {
381 let oseq: i64 = sfu_on_frame((sfuctx[0]) as *i64, smidx, rmidx, vlayer, vkey)
382 if oseq < 0 { send = 0 } else { sfu_rewrite_seq(payload, oseq) }
383 }
384 }
385 }
386 }
387 if send == 1 {
388 // frame-atomic non-blocking send (nx_sig2_txq): a slow receiver can no longer stall
389 // the room, and a torn frame can no longer poison its stream. Drops are whole-frame
390 // (seq gap -> the client's kreq recovers) and counted in tel like the old send-fails.
391 var rc: i64 = 0
392 if rcs >= 0 { rc = s2tx_send((sfuctx[8]) as *i64, rcs, pfd, fin * 256 + opcode, payload, payload_len) }
393 else { if nx_ws_send_frame_to_fd(pfd, fin, opcode, payload, payload_len) == NX_WSS_OK { rc = 1 } }
394 let ti: i64 = (ridx * NX_SIG2_MAX_PEERS + p) * 2
395 tel[ti] = tel[ti] + 1
396 if rc != 1 { tel[ti + 1] = tel[ti + 1] + 1 }
397 }
398 }
399 }
400 p = p + 1
401 }
402 return 0
403}
404
405func sig2_room_id_off(path: *u8, path_len: i64) -> i64 {
406 let prefix: *u8 = "/signal/" as *u8
407 if path_len <= 8 { return 0 - 1 }
408 var i: i64 = 0
409 while i < 8 {
410 if path[i] != prefix[i] { return 0 - 1 }
411 i = i + 1
412 }
413 return 8
414}
415
416// handle one inbound datagram. hctx: [0]=rooms [1]=id_pool [2]=id_pool_off_p [3]=members [4]=dgm_rx_mem
417// [5]=frame_out(64KB) [6]=rinfo(12 i64) [7]=packbuf(128KB) [8]=relay_seq_p [9]=now_us. Unjoined senders
418// are dropped (rule 12); a JOIN datagram (payload = room id) enrolls + ACKs; media raw-fans to UDP peers
419// and, when a frame completes reassembly, bridges as ONE WS frame to every WSS member of the room.
420// enroll a UDP sender as a room member; is_casc=1 marks it a CASCADE peer (another relay). Regular joins
421// get an ACK; cascade joins don't (relays don't wait, and the ACK would leak onto the mesh).
422func sig2_udp_enroll(ufd: i64, pkt: *u8, rinfo: *i64, from: *u8, hctx: *i64, cflags: *i64, is_casc: i64) -> i64 {
423 let plen: i64 = rinfo[5]
424 if plen < 1 { return 0 }
425 if plen > 24 { return 0 }
426 let rooms: *u8 = (hctx[0]) as *u8
427 let idp: *u8 = (hctx[1]) as *u8
428 let offp: *i64 = (hctx[2]) as *i64
429 let idptr: *u8 = ((pkt as i64) + rinfo[4]) as *u8
430 var ridx: i64 = sig2_find_room(rooms, idp, idptr, plen)
431 if ridx < 0 { ridx = sig2_alloc_room(rooms, idp, offp, idptr, plen) }
432 if ridx < 0 { return 0 }
433 let members: *u8 = (hctx[3]) as *u8
434 let mi: i64 = sig2_udpm_upsert(members, from, ridx, hctx[9])
435 if mi < 0 { return 0 }
436 cflags[mi] = is_casc
437 if is_casc == 0 {
438 let pb: *u8 = (hctx[7]) as *u8
439 let rsp: *i64 = (hctx[8]) as *i64
440 let np: i64 = dgm_pack_frame(SIG2_DG_ACK, "RELAY000" as *u8, rsp[0], "ok" as *u8, 2, pb, 131072)
441 if np == 1 {
442 rsp[0] = rsp[0] + 1
443 let pl: i64 = (pb[0] & 0xff) + ((pb[1] & 0xff) << 8)
444 sys_sendto(ufd, ((pb as i64) + 2) as *u8, pl, 0, from, 16)
445 }
446 }
447 return 0 }
448
449func sig2_udp_handle(ufd: i64, pkt: *u8, n: i64, from: *u8, hctx: *i64) -> i64 {
450 let rinfo: *i64 = (hctx[6]) as *i64
451 if dgm_parse_pkt(pkt, n, rinfo) != 0 { return 0 }
452 let dkind: i64 = rinfo[0]
453 let members: *u8 = (hctx[3]) as *u8
454 let now: i64 = hctx[9]
455 let cflags: *i64 = (hctx[10]) as *i64
456 if dkind == SIG2_DG_JOIN { return sig2_udp_enroll(ufd, pkt, rinfo, from, hctx, cflags, 0) }
457 if dkind == SIG2_DG_CASCADE { return sig2_udp_enroll(ufd, pkt, rinfo, from, hctx, cflags, 1) }
458 let mi: i64 = sig2_udpm_find(members, from)
459 if mi < 0 { return 0 }
460 let rp: *i64 = ((members as i64) + mi * 32 + 16) as *i64
461 let ridx: i64 = rp[0]
462 rp[1] = now
463 let from_cascade: i64 = cflags[mi]
464 // RAW passthrough to the other UDP members. SPLIT-HORIZON: a frame from a cascade peer goes to LOCAL
465 // members only (never back to another cascade relay) -> it crosses each inter-relay link exactly once.
466 var i: i64 = 0
467 while i < NX_SIG2_UDPM {
468 let orp: *i64 = ((members as i64) + i * 32 + 16) as *i64
469 if orp[0] == ridx { if i != mi {
470 var send_it: i64 = 1
471 if from_cascade == 1 { if cflags[i] == 1 { send_it = 0 } }
472 if send_it == 1 { sys_sendto(ufd, pkt, n, 0, ((members as i64) + i * 32) as *u8, 16) }
473 } }
474 i = i + 1
475 }
476 // BRIDGE to the WSS plane: a complete dgram frame -> one WS frame to every WSS member
477 let fout: *u8 = (hctx[5]) as *u8
478 let r: i64 = dgm_rx_add((hctx[4]) as *u8, pkt, n, fout, 65536, rinfo)
479 if r > 0 {
480 var op: i64 = 0
481 if rinfo[0] == SIG2_DG_ROOMFRAME { op = 2 }
482 if rinfo[0] == SIG2_DG_TEXT { op = 1 }
483 if op != 0 {
484 let rooms2: *u8 = (hctx[0]) as *u8
485 let rr: *i64 = sig2_room_ptr(rooms2, ridx)
486 var p2: i64 = 0
487 while p2 < NX_SIG2_MAX_PEERS {
488 let pfd: i64 = rr[R_FD0 + p2]
489 if pfd != 0 - 1 {
490 // frame-atomic non-blocking (hctx[11]=conns, [12]=txctx); a UDP burst can no
491 // longer stall the WSS plane on one slow browser
492 let bci: i64 = sig2_slot_of_fd((hctx[11]) as *u8, pfd)
493 if bci >= 0 { s2tx_send((hctx[12]) as *i64, bci, pfd, 256 + op, fout, r) }
494 else { nx_ws_send_frame_to_fd(pfd, 1, op, fout, r) }
495 }
496 p2 = p2 + 1
497 }
498 }
499 }
500 return 0 }
501
502// QOE TEE (extends the ROOMTEL sense layer, same log): a client's periodic {"type":"qoe",...} beacon
503// (fps_tx / worst fps_rx / rtt / backpressure skips / lanes up / roster-vs-tiles mismatch) is appended RAW
504// the moment it transits the relay. Fire-and-forget (Cardinal 14): a tee failure never disturbs the relay.
505// nx_health_eval aggregates -> health.json + ledger trend, so "my son had 4fps" is VISIBLE on the
506// dashboard without anyone having to report it (operator 2026-07-05: the field call had NO telemetry).
507func sig2_tee_qoe(pp: *u8, plen: i64, ridx: i64) -> i64 {
508 if plen < 12 { return 0 }
509 if plen > 512 { return 0 } // beacons are tiny; never log media-sized text
510 let ndl: *u8 = "\"type\":\"qoe\"" as *u8 // 12 bytes
511 var found: i64 = 0
512 var i: i64 = 0
513 while i + 12 <= plen {
514 var j: i64 = 0
515 var ok: i64 = 1
516 while j < 12 { if pp[i + j] != ndl[j] { ok = 0; j = 12 } else { j = j + 1 } }
517 if ok == 1 { found = 1; i = plen }
518 i = i + 1
519 }
520 if found == 0 { return 0 }
521 let lf: i64 = sys_openat_append(NX_SIG2_TEL_LOG, 420)
522 if lf < 0 { return 0 }
523 sig2_telw(lf, "QOE t=" as *u8); sig2_teln(lf, sys_now_realtime_ms() / 1000) // EPOCH (was sys_now_us = MONOTONIC -> garbage ages in the aggregator; found by the first field beacons 2026-07-05)
524 sig2_telw(lf, " room=" as *u8); sig2_teln(lf, ridx)
525 sig2_telw(lf, " " as *u8)
526 sys_write(lf, pp, plen)
527 sig2_telw(lf, "\n" as *u8)
528 sys_close(lf)
529 return 0
530}
531
532// DOMINANT-SPEAKER plane (task #41 R4). Speech scored by LPC-frame BYTE COUNT (LPC compresses silence to
533// ~nothing, speech keeps high-entropy residuals -> payload length IS an energy proxy; zero decode, zero
534// client trust). Every ~500ms per room: argmax + hysteresis (organ); on change, announce kind 0x5A
535// [dom_id 8B] to the room -- clients highlight that tile; old clients ignore unknown kinds.
536const SIG2_DOM_CHECK_US: i64 = 500000
537func sig2_audio_activity(fd: i64, ridx: i64, pp: *u8, plen: i64, sfuctx: *i64, midx: i64) -> i64 {
538 if midx < 0 { return 0 }
539 let now: i64 = sys_now_us()
540 sfu_audio_bytes((sfuctx[0]) as *i64, midx, plen, now)
541 let dom: *i64 = (sfuctx[6]) as *i64
542 let d: i64 = ridx * 2
543 if now - dom[d + 1] < SIG2_DOM_CHECK_US { return 0 }
544 dom[d + 1] = now
545 let cmx: *i64 = (sfuctx[5]) as *i64
546 let cand: *i64 = (sfuctx[7]) as *i64
547 let nc: i64 = sig2_room_cand((sfuctx[1]) as *u8, cmx, ridx, cand)
548 let nd: i64 = sfu_dominant((sfuctx[0]) as *i64, cand, nc, dom[d], now)
549 if nd == dom[d] { return 0 }
550 if nd < 0 { return 0 }
551 dom[d] = nd
552 // announce: [0x5A][dom_id 8B][seq4=0][pad] = 16B binary room frame, broadcast to EVERY member
553 let ann: *u8 = ((sfuctx[9]) + 64) as *u8
554 ann[0] = 0x5A as u8
555 let mid: *u8 = ((sfuctx[3]) + nd * 8) as *u8
556 var i: i64 = 0
557 while i < 8 { ann[1 + i] = mid[i]; i = i + 1 }
558 i = 9
559 while i < 16 { ann[i] = 0 as u8; i = i + 1 }
560 sig2_broadcast_plain(sfuctx, ridx, ann, 16)
561 return 0 }
562// plain announce to every conn in room (no SFU, no telemetry counters -- control chatter);
563// frame-atomic non-blocking via the conn's send queue (nx_sig2_txq).
564func sig2_broadcast_plain(sfuctx: *i64, ridx: i64, buf: *u8, n: i64) -> i64 {
565 let conns: *u8 = (sfuctx[1]) as *u8
566 var i: i64 = 0
567 while i < NX_SIG2_MAX_CONNS {
568 let c: *i64 = sig2_conn_ptr(conns, i)
569 if c[0] >= 0 { if c[1] == ridx { s2tx_send((sfuctx[8]) as *i64, i, c[0], 258, buf, n) } }
570 i = i + 1
571 }
572 return 0 }
573
574// LAYER SUSPENSION (task #41 R5, dynacast-class): if the LO-needed verdict for sender member smidx
575// CHANGED, tell that sender: LPROD 0x5B [id8="RELAYCTL"][seq4=0][want:1] on every conn carrying smidx.
576// Senders then produce LO only while somebody actually consumes it (battery + uplink saved).
577func sig2_lprod_eval(sfuctx: *i64, smidx: i64) -> i64 {
578 if smidx < 0 { return 0 }
579 if sfu_lprod_check((sfuctx[0]) as *i64, smidx) == 0 { return 0 }
580 let want: i64 = sfu_lprod_last((sfuctx[0]) as *i64, smidx)
581 let msg: *u8 = ((sfuctx[9]) + 128) as *u8
582 msg[0] = 0x5B as u8
583 let rid: *u8 = "RELAYCTL" as *u8
584 var i: i64 = 0
585 while i < 8 { msg[1 + i] = rid[i]; i = i + 1 }
586 i = 9
587 while i < 13 { msg[i] = 0 as u8; i = i + 1 }
588 msg[13] = want as u8
589 let conns: *u8 = (sfuctx[1]) as *u8
590 let cmx: *i64 = (sfuctx[5]) as *i64
591 var s: i64 = 0
592 while s < NX_SIG2_MAX_CONNS {
593 let c: *i64 = sig2_conn_ptr(conns, s)
594 if c[0] >= 0 { if cmx[s] == smidx { s2tx_send((sfuctx[8]) as *i64, s, c[0], 258, msg, 14) } }
595 s = s + 1
596 }
597 return 1 }
598
599// Read one frame from fd; relay/control. 0 = keep, 1 = drop conn. wuctx (R2 bridge): [0]=udp_fd
600// [1]=udp_members [2]=packbuf [3]=relay_seq_p -- every WSS frame ALSO fans out to the room's UDP members
601// (text -> SIG2_DG_TEXT, binary -> SIG2_DG_ROOMFRAME); wuctx[0]<0 disables (plane not up).
602func sig2_handle_frame(fd: i64, ridx: i64, rooms: *u8, buf: *u8, buf_cap: i64, tel: *i64, wuctx: *i64, sfuctx: *i64) -> i64 {
603 let f_raw: *u8 = ((sfuctx[9]) + 192) as *u8
604 let f: *WsFrame = f_raw as *WsFrame
605 let v: i64 = nx_ws_read_frame_from_fd(fd, buf, buf_cap, f)
606 if v != NX_WSS_OK { return 1 }
607 if f.opcode == WS_OP_CLOSE { return 1 }
608 if f.opcode == WS_OP_PING {
609 let pp: *u8 = (buf as i64 + f.payload_off) as *u8
610 nx_ws_send_pong(fd, pp, f.payload_len)
611 return 0
612 }
613 if f.opcode == WS_OP_PONG { return 0 }
614 let pp: *u8 = (buf as i64 + f.payload_off) as *u8
615 if f.opcode == WS_OP_TEXT { sig2_tee_qoe(pp, f.payload_len, ridx) } // sense layer; relay unchanged
616 if f.opcode == 2 { if f.payload_len >= 13 {
617 let mslot: i64 = sig2_slot_of_fd((sfuctx[1]) as *u8, fd)
618 if mslot >= 0 {
619 // learn this conn's member id (every binary room frame carries the sender's own id at [1..8])
620 // + resolve its MEMBER idx (the lane-stable SFU key) into the per-conn cache
621 let cid: *u8 = ((sfuctx[2]) + mslot * 8) as *u8
622 var li: i64 = 0
623 while li < 8 { cid[li] = pp[1 + li]; li = li + 1 }
624 let cmx: *i64 = (sfuctx[5]) as *i64
625 cmx[mslot] = sig2_memidx((sfuctx[0]) as *i64, (sfuctx[3]) as *u8, (sfuctx[4]) as *i64, cid)
626 let fkind: i64 = pp[0] & 0xff
627 if fkind == 0x4C { sig2_audio_activity(fd, ridx, pp, f.payload_len, sfuctx, cmx[mslot]) }
628 if fkind == 0x41 { sig2_audio_activity(fd, ridx, pp, f.payload_len, sfuctx, cmx[mslot]) }
629 if fkind == 0x6C { sig2_audio_activity(fd, ridx, pp, f.payload_len, sfuctx, cmx[mslot]) } // E2EE LPC: length-based
630 if fkind == 0x61 { sig2_audio_activity(fd, ridx, pp, f.payload_len, sfuctx, cmx[mslot]) } // E2EE PCM: energy still works
631 if fkind == 0x52 {
632 // RECEIVER REPORT (TWCC-class, ladder feedback): [target8 at 13..20][rxfps:1 at 21]...
633 // routed to the TARGET SENDER's conns ONLY (never broadcast) -- the sender's ladder learns
634 // what each receiver ACTUALLY gets (sender-blindness was the last congestion gap).
635 if f.payload_len >= 22 {
636 let conns3: *u8 = (sfuctx[1]) as *u8
637 var rr_i: i64 = 0
638 while rr_i < NX_SIG2_MAX_CONNS {
639 let c3: *i64 = sig2_conn_ptr(conns3, rr_i)
640 if c3[0] >= 0 { if c3[1] == ridx { if rr_i != mslot {
641 let tcid3: *u8 = ((sfuctx[2]) + rr_i * 8) as *u8
642 var eq3: i64 = 1
643 var q3: i64 = 0
644 while q3 < 8 { if tcid3[q3] != pp[13 + q3] { eq3 = 0; q3 = 8 } else { q3 = q3 + 1 } }
645 if eq3 == 1 { s2tx_send((sfuctx[8]) as *i64, rr_i, c3[0], 258, pp, f.payload_len) }
646 } } }
647 rr_i = rr_i + 1
648 }
649 }
650 return 0
651 }
652 if fkind == 0x59 {
653 // LNSET (task #41 R4): receiver's last-N cap [n:1], device-derived by the client
654 // (measured decode capability), NOT a server-hardcoded number. Consumed, never broadcast.
655 if f.payload_len >= 14 { sfu_set_lastn((sfuctx[0]) as *i64, cmx[mslot], pp[13] & 0xff) }
656 return 0
657 }
658 if fkind == 0x58 {
659 // LSUB (task #41): receiver->relay layer subscription. CONSUMED here -- never broadcast,
660 // never bridged. payload = [target_id 8B][want 1B]; target must be a member of THIS room
661 // (some conn in the room has learned that id).
662 let tid: *u8 = ((sfuctx[9]) + 384) as *u8
663 let want: i64 = sfu_parse_lsub(pp, f.payload_len, tid)
664 if want >= 0 {
665 let conns2: *u8 = (sfuctx[1]) as *u8
666 var ts: i64 = 0
667 while ts < NX_SIG2_MAX_CONNS {
668 let c2: *i64 = sig2_conn_ptr(conns2, ts)
669 if c2[0] >= 0 { if c2[1] == ridx { if ts != mslot {
670 let tcid: *u8 = ((sfuctx[2]) + ts * 8) as *u8
671 var eq: i64 = 1
672 var q: i64 = 0
673 while q < 8 { if tcid[q] != tid[q] { eq = 0; q = 8 } else { q = q + 1 } }
674 if eq == 1 {
675 let tmidx: i64 = sig2_memidx((sfuctx[0]) as *i64, (sfuctx[3]) as *u8, (sfuctx[4]) as *i64, tid)
676 sfu_want((sfuctx[0]) as *i64, tmidx, cmx[mslot], want)
677 sig2_lprod_eval(sfuctx, tmidx) // R5: tell the target sender if LO-need flipped
678 ts = NX_SIG2_MAX_CONNS
679 }
680 } } }
681 if ts < NX_SIG2_MAX_CONNS { ts = ts + 1 }
682 }
683 }
684 return 0
685 }
686 }
687 } }
688 sig2_broadcast(rooms, ridx, fd, f.fin, f.opcode, pp, f.payload_len, tel, sfuctx)
689 if wuctx[0] >= 0 {
690 var dk: i64 = 0
691 if f.opcode == WS_OP_TEXT { dk = SIG2_DG_TEXT }
692 if f.opcode == 2 { dk = SIG2_DG_ROOMFRAME }
693 if dk != 0 { sig2_udp_fanout(wuctx[0], (wuctx[1]) as *u8, ridx, 0 - 1, dk, pp, f.payload_len,
694 (wuctx[2]) as *u8, (wuctx[3]) as *i64) }
695 }
696 return 0
697}
698
699// EDGE side of the cascade: announce CASCADE-JOIN(room) to HOME + register HOME as a cascade member here
700// (so HOME's frames split-horizon on this relay too). Called at startup + periodically (keepalive).
701func sig2_cascade_keepalive(ufd: i64, home_addr: *u8, room: *u8, roomlen: i64, hctx: *i64, cflags: *i64) -> i64 {
702 let pb: *u8 = (hctx[7]) as *u8
703 let rsp: *i64 = (hctx[8]) as *i64
704 let np: i64 = dgm_pack_frame(SIG2_DG_CASCADE, "RELAYEDG" as *u8, rsp[0], room, roomlen, pb, 131072)
705 if np == 1 {
706 rsp[0] = rsp[0] + 1
707 let pl: i64 = (pb[0] & 0xff) + ((pb[1] & 0xff) << 8)
708 sys_sendto(ufd, ((pb as i64) + 2) as *u8, pl, 0, home_addr, 16)
709 }
710 let rooms: *u8 = (hctx[0]) as *u8
711 let idp: *u8 = (hctx[1]) as *u8
712 let offp: *i64 = (hctx[2]) as *i64
713 var ridx: i64 = sig2_find_room(rooms, idp, room, roomlen)
714 if ridx < 0 { ridx = sig2_alloc_room(rooms, idp, offp, room, roomlen) }
715 if ridx >= 0 {
716 let members: *u8 = (hctx[3]) as *u8
717 let mi: i64 = sig2_udpm_upsert(members, home_addr, ridx, hctx[9])
718 if mi >= 0 { cflags[mi] = 1 }
719 }
720 return 0 }
721
722func main(argc: i64, argv: *i64) -> i64 {
723 // THE 2026-07-05 OUTAGE FIX (supervisor reap: sig=13 crash-loop): a phone dropping mid-relay left a
724 // dead socket; the next broadcast write raised SIGPIPE and KILLED THE WHOLE DAEMON -- every lane in
725 // every room dropped ("relay: reconnecting..."), chronically, for every call. SIG_IGN makes that
726 // write return -EPIPE instead; sig2_broadcast already counts it (tel send_fails) and the dead peer's
727 // read path closes it. One dying phone must never take down the room again -- by construction.
728 nx_signal_ignore(NX_SIGPIPE)
729 // ---- CASCADE / multi-instance args (Octo-class). Default (no args) = today's single relay EXACTLY.
730 // argv[3]=wss_port argv[4]=udp_port argv[5]=cascade_home_ip argv[6]=cascade_home_port argv[7]=cascade_room
731 var wss_port: i64 = NX_SIG2_PORT
732 var udp_port: i64 = NX_SIG2_UDP_PORT
733 if argc > 3 { let v: i64 = sig2_atoi((argv[3]) as *u8); if v > 0 { wss_port = v } }
734 if argc > 4 { let v: i64 = sig2_atoi((argv[4]) as *u8); if v > 0 { udp_port = v } }
735 var casc_on: i64 = 0
736 let casc_home: *u8 = sys_mmap(16)
737 let casc_room: *u8 = sys_mmap(32)
738 var casc_roomlen: i64 = 0
739 if argc > 7 {
740 casc_home[0] = 2 as u8; casc_home[1] = 0 as u8
741 let hp: i64 = sig2_atoi((argv[6]) as *u8)
742 casc_home[2] = ((hp >> 8) & 0xff) as u8; casc_home[3] = (hp & 0xff) as u8
743 sig2_parse_ip((argv[5]) as *u8, ((casc_home as i64) + 4) as *u8)
744 var cz: i64 = 8; while cz < 16 { casc_home[cz] = 0 as u8; cz = cz + 1 }
745 let rr: *u8 = (argv[7]) as *u8
746 var rl: i64 = 0; while rr[rl] != (0 as u8) { casc_room[rl] = rr[rl]; rl = rl + 1 }
747 casc_roomlen = rl
748 casc_on = 1
749 }
750 let sfd: i64 = sys_socket(AF_INET, SOCK_STREAM, 0)
751 if sfd < 0 { sys_exit(101); return 101 }
752 let opt: *u8 = sys_mmap(8)
753 opt[0] = 1 as u8; opt[1] = 0 as u8; opt[2] = 0 as u8; opt[3] = 0 as u8
754 sys_setsockopt(sfd, SOL_SOCKET, SO_REUSEADDR, opt, 4)
755 let addr: *u8 = sys_mmap(16)
756 sig2_sockaddr_loopback(addr, wss_port)
757 if sys_bind(sfd, addr, 16) < 0 { sys_exit(102); return 102 }
758 if sys_listen(sfd, NX_SIG2_ACCEPT_BACKLOG) < 0 { sys_exit(103); return 103 }
759 sys_write(2, "nx_signaling_v2 on 127.0.0.1:8445 (N-party rooms, RFC 6455)\n" as *u8, 61)
760
761 let rooms: *u8 = sys_mmap(NX_SIG2_MAX_ROOMS * NX_SIG2_ROOM_BYTES)
762 let id_pool: *u8 = sys_mmap(NX_SIG2_ROOM_ID_POOL_BYTES)
763 let id_pool_off_p: *i64 = sys_mmap(16) as *i64
764 id_pool_off_p[0] = 0
765
766 let conns: *u8 = sys_mmap(NX_SIG2_MAX_CONNS * NX_SIG2_CONN_BYTES)
767 var i: i64 = 0
768 while i < NX_SIG2_MAX_CONNS {
769 let c: *i64 = sig2_conn_ptr(conns, i)
770 c[0] = 0 - 1 // fd
771 c[1] = 0 - 1 // room_idx
772 i = i + 1
773 }
774
775 // ---- R2: the SOVEREIGN UDP plane socket (:8471, INADDR_ANY -- native clients hit it directly) ----
776 let ufd: i64 = sys_socket(AF_INET, SOCK_DGRAM, 0)
777 var udp_up: i64 = 0 - 1
778 let uaddr0: *u8 = sys_mmap(16)
779 uaddr0[0] = 2 as u8; uaddr0[1] = 0 as u8
780 uaddr0[2] = ((udp_port >> 8) & 0xff) as u8
781 uaddr0[3] = (udp_port & 0xff) as u8
782 var uz: i64 = 4
783 while uz < 16 { uaddr0[uz] = 0 as u8; uz = uz + 1 }
784 if ufd >= 0 { if sys_bind(ufd, uaddr0, 16) >= 0 { udp_up = ufd
785 sys_write(2, "sig2: UDP plane on 0.0.0.0:8471 (sovereign dgram)\n" as *u8, 50) } }
786 let udpm: *u8 = sys_mmap(NX_SIG2_UDPM * 32)
787 var um: i64 = 0
788 while um < NX_SIG2_UDPM { let urp: *i64 = ((udpm as i64) + um * 32 + 16) as *i64; urp[0] = 0 - 1; um = um + 1 }
789 let dgmem: *u8 = sys_mmap(DGM_RX_REGION)
790 dgm_rx_init(dgmem)
791 let ufout: *u8 = sys_mmap(65536)
792 let urinfo: *i64 = sys_mmap(8 * 12) as *i64
793 let upack: *u8 = sys_mmap(131072)
794 let urseq: *i64 = sys_mmap(16) as *i64
795 urseq[0] = 1
796 let urecv: *u8 = sys_mmap(2048)
797 let ufrom: *u8 = sys_mmap(16)
798 let ufromlen: *i64 = sys_mmap(16) as *i64
799 let cascade_flags: *i64 = sys_mmap(8 * NX_SIG2_UDPM) as *i64 // per UDP member: 1 = a cascade relay peer
800 var cf: i64 = 0
801 while cf < NX_SIG2_UDPM { cascade_flags[cf] = 0; cf = cf + 1 }
802 let hctx: *i64 = sys_mmap(8 * 13) as *i64
803 hctx[0] = rooms as i64; hctx[1] = id_pool as i64; hctx[2] = id_pool_off_p as i64
804 hctx[3] = udpm as i64; hctx[4] = dgmem as i64; hctx[5] = ufout as i64
805 hctx[6] = urinfo as i64; hctx[7] = upack as i64; hctx[8] = urseq as i64; hctx[9] = 0
806 hctx[10] = cascade_flags as i64 // [11]=conns [12]=txctx set below once allocated (UDP->WSS bridge sends)
807 let wuctx: *i64 = sys_mmap(8 * 4) as *i64
808 wuctx[0] = udp_up; wuctx[1] = udpm as i64; wuctx[2] = upack as i64; wuctx[3] = urseq as i64
809
810 // ---- SFU (task #41): selective-forwarding state, keyed by MEMBER id (lane/reconnect-stable) ----
811 let sfu_mem: *i64 = sys_mmap(SFU_REGION) as *i64
812 sfu_init(sfu_mem)
813 let conn_ids: *u8 = sys_mmap(NX_SIG2_MAX_CONNS * 8) // member id (8B) learned per conn slot
814 let mem_ids: *u8 = sys_mmap(SFU_MAXC * 8) // member registry: idx -> id
815 let mcount_p: *i64 = sys_mmap(16) as *i64
816 mcount_p[0] = 0
817 let conn_memidx: *i64 = sys_mmap(8 * NX_SIG2_MAX_CONNS) as *i64
818 var cmi: i64 = 0
819 while cmi < NX_SIG2_MAX_CONNS { conn_memidx[cmi] = 0 - 1; cmi = cmi + 1 }
820 // ---- ROOM TOKENS (capability links): secret + protected list are FILES (data-driven; argv override
821 // for the gate). No secret OR empty list -> every room open = exactly today's behavior. ----
822 var rk_sec_path: *u8 = "knowledge/room_secret.txt" as *u8
823 var rk_lst_path: *u8 = "knowledge/rooms_protected.txt" as *u8
824 if argc > 1 { rk_sec_path = (argv[1]) as *u8 }
825 if argc > 2 { rk_lst_path = (argv[2]) as *u8 }
826 let rk_sec_len_p: *i64 = sys_mmap(16) as *i64
827 var rk_sec: *u8 = sys_read_file(rk_sec_path, rk_sec_len_p)
828 var rk_sec_len: i64 = 0
829 if (rk_sec as i64) != 0 {
830 rk_sec_len = rk_sec_len_p[0]
831 var trimming: i64 = 1 // trim trailing CR/LF so editors can't break tokens
832 while trimming == 1 {
833 if rk_sec_len <= 0 { trimming = 0 }
834 else {
835 let lc: i64 = rk_sec[rk_sec_len - 1] & 0xff
836 if lc == 10 { rk_sec_len = rk_sec_len - 1 }
837 else { if lc == 13 { rk_sec_len = rk_sec_len - 1 } else { trimming = 0 } }
838 }
839 }
840 }
841 let rk_lst_len_p: *i64 = sys_mmap(16) as *i64
842 var rk_lst: *u8 = sys_read_file(rk_lst_path, rk_lst_len_p)
843 var rk_lst_len: i64 = 0
844 if (rk_lst as i64) != 0 { rk_lst_len = rk_lst_len_p[0] }
845 if rk_sec_len > 0 { if rk_lst_len > 0 { sys_write(2, "sig2: room-token wall armed (protected list present)\n" as *u8, 53) } }
846
847 let dom_tab: *i64 = sys_mmap(8 * 2 * NX_SIG2_MAX_ROOMS) as *i64 // per room: [dom_midx, last_check_us]
848 var dmi: i64 = 0
849 while dmi < NX_SIG2_MAX_ROOMS { dom_tab[dmi*2] = 0 - 1; dom_tab[dmi*2+1] = 0; dmi = dmi + 1 }
850 let cand_scr: *i64 = sys_mmap(8 * NX_SIG2_MAX_CONNS) as *i64 // room-candidate scratch (single-threaded)
851 // frame-atomic TX queues (nx_sig2_txq): one 256KB queue per conn slot + head/len + a header scratch.
852 // The slab is mmap-virtual -- pages are touched only when a receiver actually backpressures.
853 let txq: *u8 = sys_mmap(NX_SIG2_MAX_CONNS * NX_S2TX_QBYTES)
854 let txh: *i64 = sys_mmap(8 * NX_SIG2_MAX_CONNS) as *i64
855 let txl: *i64 = sys_mmap(8 * NX_SIG2_MAX_CONNS) as *i64
856 let txhscr: *u8 = sys_mmap(16)
857 let txctx: *i64 = sys_mmap(8 * 4) as *i64
858 txctx[0] = txq as i64; txctx[1] = txh as i64; txctx[2] = txl as i64; txctx[3] = txhscr as i64
859 hctx[11] = conns as i64; hctx[12] = txctx as i64 // UDP->WSS bridge sends go frame-atomic too
860 let sfuctx: *i64 = sys_mmap(8 * 12) as *i64
861 sfuctx[0] = sfu_mem as i64; sfuctx[1] = conns as i64; sfuctx[2] = conn_ids as i64
862 sfuctx[3] = mem_ids as i64; sfuctx[4] = mcount_p as i64; sfuctx[5] = conn_memidx as i64
863 sfuctx[6] = dom_tab as i64; sfuctx[7] = cand_scr as i64
864 sfuctx[8] = txctx as i64 // send-queue ctx: every room-frame send goes through s2tx_send
865 // seq1134 LEAK FIX (2026-07-28): the frame/relay hot paths called sys_mmap PER EVENT -- with no
866 // munmap in this runtime and page-granular maps, ~40 events/sec leaked into the 2.9GB RSS wedge
867 // that silently killed the family call (daemon up + listening + relaying nothing). ONE startup
868 // region, fixed offsets reused per event; the poll loop is single-threaded, so reuse is race-free
869 // by construction. Layout: +0 vinf(32) +64 ann(32) +128 msg(32) +192 fraw(128) +384 tid(16).
870 sfuctx[9] = sys_mmap(512) as i64
871
872 let n_pollfds: i64 = NX_SIG2_MAX_CONNS + 2
873 let pfds: *u8 = sys_mmap(n_pollfds * NX_POLLFD_BYTES)
874 nx_pollfd_set(pfds, 0, sfd, NX_POLLIN)
875 var pi: i64 = 1
876 while pi < n_pollfds { nx_pollfd_set(pfds, pi, 0 - 1, NX_POLLIN); pi = pi + 1 }
877 if udp_up >= 0 { nx_pollfd_set(pfds, NX_SIG2_MAX_CONNS + 1, udp_up, NX_POLLIN) }
878
879 let frame_buf: *u8 = sys_mmap(NX_SIG2_FRAME_BUF_BYTES)
880 let hdr_buf: *u8 = sys_mmap(NX_SIG2_HDR_BUF_BYTES)
881 let path_buf: *u8 = sys_mmap(256)
882 let key_buf: *u8 = sys_mmap(128)
883 let path_n_p: *i64 = sys_mmap(16) as *i64
884 let key_n_p: *i64 = sys_mmap(16) as *i64
885
886 let tel: *i64 = sys_mmap(8 * NX_SIG2_TEL_SLOTS) as *i64 // per (room,peer) relayed/fails, zero-init
887 var last_emit: i64 = sys_now_us()
888 // CASCADE edge: establish the link to HOME immediately (don't wait a full tick)
889 if casc_on == 1 { if udp_up >= 0 {
890 hctx[9] = sys_now_us()
891 sig2_cascade_keepalive(udp_up, casc_home, casc_room, casc_roomlen, hctx, cascade_flags)
892 sys_write(2, "sig2: CASCADE edge -> HOME (Octo-class inter-relay link)\n" as *u8, 56)
893 } }
894
895 var keep: i64 = 1
896 while keep == 1 {
897 let now0: i64 = sys_now_us()
898 if now0 - last_emit > NX_SIG2_EMIT_US {
899 sig2_emit_tel(rooms, tel, now0); last_emit = now0
900 var sw: i64 = 0 // R2: sweep UDP members idle >30s (dead NAT bindings)
901 while sw < NX_SIG2_UDPM {
902 let srp: *i64 = ((udpm as i64) + sw * 32 + 16) as *i64
903 if srp[0] >= 0 { if now0 - srp[1] > 30000000 { srp[0] = 0 - 1 } }
904 sw = sw + 1
905 }
906 if casc_on == 1 { if udp_up >= 0 { // cascade keepalive: refresh the link (both directions) < the 30s sweep
907 hctx[9] = now0
908 sig2_cascade_keepalive(udp_up, casc_home, casc_room, casc_roomlen, hctx, cascade_flags)
909 } }
910 }
911 // TX-QUEUE event sweep: a conn with queued bytes also polls POLLOUT so its queue drains the
912 // moment the socket has room (frame-atomic continuation -- see nx_sig2_txq).
913 var pm: i64 = 0
914 while pm < NX_SIG2_MAX_CONNS {
915 let cpm: *i64 = sig2_conn_ptr(conns, pm)
916 if cpm[0] != 0 - 1 {
917 var pev: i64 = NX_POLLIN
918 if txl[pm] > 0 { pev = NX_POLLIN | NX_POLLOUT }
919 nx_pollfd_set(pfds, pm + 1, cpm[0], pev)
920 }
921 pm = pm + 1
922 }
923 // 5s tick, NOT block-forever: nx_poll(-1) lowers to ppoll with a
924 // NULL timespec, and that branch miscompiles under the C-bootstrap
925 // compiler into a ZERO timespec -> instant return -> 100% CPU spin
926 // (found live on the NAS 2026-06-10). A positive timeout uses the
927 // proven path; 0.2 wakeups/sec is negligible.
928 let n_ready: i64 = nx_poll(pfds, n_pollfds, 5000)
929 if n_ready < 0 { keep = 0 }
930 if n_ready == 0 { continue }
931
932 if (nx_pollfd_revents(pfds, 0) & NX_POLLIN) != 0 {
933 let cfd: i64 = sys_accept(sfd)
934 if cfd >= 0 {
935 // TCP_NODELAY (IPPROTO_TCP=6, TCP_NODELAY=1): the relay's per-peer sockets carry
936 // every broadcast frame back toward the pumps. Nagle here x delayed-ACK on the
937 // pump side = the ~40ms burst floor nx_video_qoe_live measured 2026-07-03.
938 let snd: *u8 = sys_mmap(4)
939 snd[0] = 1 as u8
940 snd[1] = 0 as u8
941 snd[2] = 0 as u8
942 snd[3] = 0 as u8
943 sys_setsockopt(cfd, 6, 1, snd, 4)
944 let v: i64 = nx_ws_upgrade_handshake(cfd,
945 hdr_buf, NX_SIG2_HDR_BUF_BYTES,
946 path_buf, 256, path_n_p,
947 key_buf, 128, key_n_p)
948 if v != NX_WSU_OK {
949 // Cardinal 18: say WHAT failed so a dead join is debuggable.
950 sys_write(2, "sig2: upgrade rejected verdict=" as *u8, 31)
951 let vb: *u8 = sys_mmap(8)
952 vb[0] = (48 + v) as u8
953 vb[1] = 10 as u8
954 sys_write(2, vb, 2)
955 sys_close(cfd)
956 } else {
957 let room_off: i64 = sig2_room_id_off(path_buf, path_n_p[0])
958 if room_off < 0 {
959 nx_ws_send_close(cfd, 1008)
960 sys_close(cfd)
961 } else {
962 let id_ptr: *u8 = (path_buf as i64 + room_off) as *u8
963 // ROOM IDENTITY excludes any ?k=... query (a keyed join must land in the SAME room)
964 let id_len: i64 = rk_path_no_query(id_ptr, path_n_p[0] - room_off)
965 // TOKEN WALL: protected base room -> the ?k= token must equal sha256(base||secret).
966 // Empty list / no secret = every room open (today's behavior by construction).
967 var admit: i64 = 1
968 if rk_sec_len > 0 { if rk_lst_len > 0 {
969 let blen: i64 = rk_base_len(id_ptr, id_len)
970 if rk_protected(rk_lst, rk_lst_len, id_ptr, blen) == 1 {
971 admit = 0
972 let kbuf: *u8 = sys_mmap(64)
973 let klen: i64 = rk_parse_k(path_buf, path_n_p[0], kbuf, 32)
974 if klen == RK_TOK_LEN {
975 let want: *u8 = sys_mmap(32)
976 rk_token(id_ptr, blen, rk_sec, rk_sec_len, want)
977 var eqk: i64 = 1
978 var qk: i64 = 0
979 while qk < RK_TOK_LEN { if kbuf[qk] != want[qk] { eqk = 0; qk = RK_TOK_LEN } else { qk = qk + 1 } }
980 if eqk == 1 { admit = 1 }
981 }
982 }
983 } }
984 if admit == 0 {
985 nx_ws_send_close(cfd, 4003) // keyed room: token absent or wrong
986 sys_close(cfd)
987 } else {
988 var ridx: i64 = sig2_find_room(rooms, id_pool, id_ptr, id_len)
989 if ridx < 0 {
990 ridx = sig2_alloc_room(rooms, id_pool, id_pool_off_p, id_ptr, id_len)
991 }
992 if ridx < 0 {
993 nx_ws_send_close(cfd, 1011)
994 sys_close(cfd)
995 } else {
996 let slot: i64 = sig2_join(rooms, ridx, cfd)
997 if slot < 0 {
998 nx_ws_send_close(cfd, 1008) // room full (8 peers)
999 sys_close(cfd)
1000 } else {
1001 var ci: i64 = 0
1002 var placed: i64 = 0
1003 while ci < NX_SIG2_MAX_CONNS {
1004 let c: *i64 = sig2_conn_ptr(conns, ci)
1005 if c[0] == 0 - 1 {
1006 if placed == 0 {
1007 c[0] = cfd
1008 c[1] = ridx
1009 txh[ci] = 0 // fresh conn slot -> empty send queue
1010 txl[ci] = 0
1011 nx_pollfd_set(pfds, ci + 1, cfd, NX_POLLIN)
1012 placed = 1
1013 }
1014 }
1015 ci = ci + 1
1016 }
1017 if placed == 0 {
1018 sig2_leave(rooms, ridx, cfd)
1019 sys_close(cfd)
1020 }
1021 }
1022 }
1023 }
1024 }
1025 }
1026 }
1027 }
1028
1029 // R2: drain one datagram per wake (poll re-fires if more are queued -- no nonblock needed)
1030 if udp_up >= 0 {
1031 if (nx_pollfd_revents(pfds, NX_SIG2_MAX_CONNS + 1) & NX_POLLIN) != 0 {
1032 ufromlen[0] = 16
1033 let rn: i64 = sys_recvfrom(udp_up, urecv, 2048, 0, ufrom, ufromlen)
1034 if rn > 0 { hctx[9] = now0; sig2_udp_handle(udp_up, urecv, rn, ufrom, hctx) }
1035 }
1036 }
1037
1038 var ci: i64 = 0
1039 while ci < NX_SIG2_MAX_CONNS {
1040 let c: *i64 = sig2_conn_ptr(conns, ci)
1041 if c[0] != 0 - 1 {
1042 let rev: i64 = nx_pollfd_revents(pfds, ci + 1)
1043 if rev != 0 {
1044 var dropv: i64 = 0
1045 if (rev & NX_POLLOUT) != 0 { // socket has room -> continue the queued bytes
1046 if s2tx_drain(txq, txh, txl, ci, c[0]) < 0 { dropv = 1 }
1047 }
1048 if dropv == 0 {
1049 if (rev & (NX_POLLIN | NX_POLLERR | NX_POLLHUP | NX_POLLNVAL)) != 0 {
1050 dropv = sig2_handle_frame(c[0], c[1], rooms, frame_buf, NX_SIG2_FRAME_BUF_BYTES, tel, wuctx, sfuctx)
1051 }
1052 }
1053 if dropv == 1 {
1054 sig2_leave(rooms, c[1], c[0])
1055 sys_close(c[0])
1056 c[0] = 0 - 1
1057 c[1] = 0 - 1
1058 conn_memidx[ci] = 0 - 1 // conn slot free; MEMBER pairs persist (lane
1059 txh[ci] = 0 // hops + reconnects keep seq monotonic)
1060 txl[ci] = 0
1061 nx_pollfd_set(pfds, ci + 1, 0 - 1, NX_POLLIN)
1062 }
1063 }
1064 }
1065 ci = ci + 1
1066 }
1067 }
1068 sys_close(sfd)
1069 return 0
1070}