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}