code wiki / _hdl_build / nx_signaling_v2.nx
`nx_signaling_v2.nx` source

↩ module page · 1081 lines · 54722 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 {
  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 {
  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 {
  addr[0] = 2 as u8; addr[1] = 0 as u8
  addr[2] = ((port >> 8) & 0xff) as u8
  addr[3] = (port & 0xff) as u8
  addr[4] = 127 as u8; addr[5] = 0 as u8; addr[6] = 0 as u8; addr[7] = 1 as u8
  addr[8]  = 0 as u8; addr[9]  = 0 as u8; addr[10] = 0 as u8; addr[11] = 0 as u8
  addr[12] = 0 as u8; addr[13] = 0 as u8; addr[14] = 0 as u8; addr[15] = 0 as u8
  return 16
74}
75
76func sig2_find_room(rooms: *u8, id_pool: *u8, id: *u8, id_len: i64) -> i64 {
  var i: i64 = 0
  while i < NX_SIG2_MAX_ROOMS {
      let r: *i64 = sig2_room_ptr(rooms, i)
      if r[R_USED] == 1 {
          if r[R_ID_LEN] == id_len {
              var j: i64 = 0
              var ok: i64 = 1
              while j < id_len {
                  if id_pool[r[R_ID_OFF] + j] != id[j] { ok = 0; j = id_len } else { j = j + 1 }
              }
              if ok == 1 { return i }
          }
      }
      i = i + 1
  }
  return 0 - 1
93}
94
95// FIXED PER-SLOT ID REGION (2026-09-02, measured live): the id pool used to be a BUMP allocator -- every new room
96// name was appended at id_pool_off and sig2_leave freed the ROOM SLOT but never the NAME BYTES, so after ~4096 B
97// of DISTINCT room names since process start every NEW room was refused with close 1011 FOREVER while existing
98// rooms kept working (two probe lanes x three WS lanes each, all 1011, for hours; the family could not open a new
99// room name). Slot i now owns bytes [i*SLOT, (i+1)*SLOT) of the SAME pool, so a freed slot's bytes are reused by
100// construction. The pool size is unchanged and SLOT is DERIVED from it (no new constant). -1 = no free room slot,
101// -2 = the name is longer than a slot (the client caps names at 24 chars plus a lane suffix, far below SLOT).
102func sig2_alloc_room(rooms: *u8, id_pool: *u8, id_pool_off_p: *i64, id: *u8, id_len: i64) -> i64 {
  let slot_bytes: i64 = NX_SIG2_ROOM_ID_POOL_BYTES / NX_SIG2_MAX_ROOMS
  if id_len > slot_bytes { return 0 - 2 }
  var i: i64 = 0
  while i < NX_SIG2_MAX_ROOMS {
      let r: *i64 = sig2_room_ptr(rooms, i)
      if r[R_USED] == 0 {
          let off: i64 = i * slot_bytes
          var k: i64 = 0
          while k < id_len { id_pool[off + k] = id[k]; k = k + 1 }
          id_pool_off_p[0] = off + id_len     // end of the LAST claim -- observability only, no longer a cursor
          r[R_ID_OFF] = off
          r[R_ID_LEN] = id_len
          r[R_USED] = 1
          r[R_NPEERS] = 0
          var p: i64 = 0
          while p < NX_SIG2_MAX_PEERS { r[R_FD0 + p] = 0 - 1; p = p + 1 }
          return i
      }
      i = i + 1
  }
  return 0 - 1
124}
125
126// Join: first free peer slot.  Returns slot or -1 (room full).
127func sig2_join(rooms: *u8, ridx: i64, fd: i64) -> i64 {
  let r: *i64 = sig2_room_ptr(rooms, ridx)
  var p: i64 = 0
  while p < NX_SIG2_MAX_PEERS {
      if r[R_FD0 + p] == 0 - 1 {
          r[R_FD0 + p] = fd
          r[R_NPEERS] = r[R_NPEERS] + 1
          return p
      }
      p = p + 1
  }
  return 0 - 1
139}
140
141// Leave: clear fd's slot; free the room when empty.
142func sig2_leave(rooms: *u8, ridx: i64, fd: i64) -> i64 {
  let r: *i64 = sig2_room_ptr(rooms, ridx)
  var p: i64 = 0
  while p < NX_SIG2_MAX_PEERS {
      if r[R_FD0 + p] == fd {
          r[R_FD0 + p] = 0 - 1
          r[R_NPEERS] = r[R_NPEERS] - 1
      }
      p = p + 1
  }
  if r[R_NPEERS] <= 0 { r[R_USED] = 0; r[R_NPEERS] = 0 }
  return 0
154}
155
156// --- per-participant relay telemetry (the "where does it suck" SENSE layer) ---
157// tel[] = per (room,peer): [frames_relayed, send_fails].  A peer with rising send_fails
158// is the slow/wedged reader = the bottleneck participant (room-perf-arc: "relay wedges
159// when a peer's buffer fills").  Fire-and-forget (Cardinal 14): a logging failure must
160// NEVER disturb the relay.  Emitted every NX_SIG2_EMIT_US by the main loop.
161const NX_SIG2_TEL_LOG: *u8 = "/volume1/homes/elderwesto/nishihost/room_telemetry.log"
162const NX_SIG2_EMIT_US: i64 = 5000000
163const NX_SIG2_TEL_SLOTS: i64 = 1024   // MAX_ROOMS(64) * MAX_PEERS(8) * 2
164
165// ---- R2 SOVEREIGN UDP PLANE (task #15; operator: pure nishi, hardware rung up, NO WebTransport) ----
166// Native clients (#22) + NishiOS speak nx_dgram_media datagrams straight to :8471; browser tabs stay on
167// the WSS lane (their sandbox has no raw UDP). SAME room registry, MIXED-PLANE rooms: the relay BRIDGES --
168// UDP->WSS: reassemble a dgram frame -> emit as one WS frame to every WSS member; WSS->UDP: fragment each
169// WS frame -> datagrams to every UDP member. The dgram payload IS the room-protocol frame (protocol SSOT
170// #21 unchanged); the dgram header is transport-only. JOIN: kind 0x4A datagram, payload = room id; the
171// client re-JOINs every ~5s (presence + NAT keepalive); members idle >30s are swept.
172const NX_SIG2_UDP_PORT: i64 = 8471
173const NX_SIG2_UDPM: i64 = 16          // UDP members (addr16 | room:i64 | last_us:i64 = 32B each)
174const SIG2_DG_JOIN: i64 = 0x4A
175const SIG2_DG_ACK:  i64 = 0x4B
176const SIG2_DG_ROOMFRAME: i64 = 0x4D   // payload = a room-protocol BINARY frame (bridged <-> WS binary)
177const SIG2_DG_TEXT: i64 = 0x54        // payload = a room-protocol TEXT frame (bridged <-> WS text JSON)
178const SIG2_DG_CASCADE: i64 = 0x4E     // CASCADE-JOIN (relay-to-relay, Octo-class): enroll the sender as a
179//                                        CASCADE member -> a frame from a cascade member fans to LOCAL peers
180//                                        ONLY (split-horizon) so it crosses each inter-relay link exactly ONCE.
181
182func sig2_atoi(s: *u8) -> i64 {
  var v: i64 = 0; var i: i64 = 0
  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 }
  return v }
186// parse dotted-decimal "a.b.c.d" -> out[0..3]
187func sig2_parse_ip(s: *u8, out: *u8) -> i64 {
  var oi: i64 = 0; var v: i64 = 0; var i: i64 = 0; var run: i64 = 1
  while run == 1 {
      let c: i64 = s[i] & 0xff
      if c == 0 { if oi < 4 { out[oi] = v as u8 } run = 0 }
      else { if c == 46 { if oi < 4 { out[oi] = v as u8 } oi = oi + 1; v = 0 }
             else { if c >= 48 { if c <= 57 { v = v * 10 + (c - 48) } } } }
      i = i + 1
      if i > 24 { run = 0 }
  }
  return 0 }
198
199func sig2_udpm_find(m: *u8, addr: *u8) -> i64 {
  var i: i64 = 0
  while i < NX_SIG2_UDPM {
      let b: *u8 = ((m as i64) + i * 32) as *u8
      let rp: *i64 = ((m as i64) + i * 32 + 16) as *i64
      if rp[0] >= 0 {
          var eq: i64 = 1
          var k: i64 = 0
          while k < 8 { if b[k] != addr[k] { eq = 0; k = 8 } else { k = k + 1 } }
          if eq == 1 { return i }
      }
      i = i + 1
  }
  return 0 - 1 }
213func sig2_udpm_upsert(m: *u8, addr: *u8, ridx: i64, now: i64) -> i64 {
  var i: i64 = sig2_udpm_find(m, addr)
  if i < 0 {
      var j: i64 = 0
      while j < NX_SIG2_UDPM {
          if i < 0 {
              let rp: *i64 = ((m as i64) + j * 32 + 16) as *i64
              if rp[0] < 0 { i = j }
              if rp[0] >= 0 { if now - rp[1] > 30000000 { i = j } }   // sweep stale >30s
          }
          j = j + 1
      }
  }
  if i < 0 { return 0 - 1 }
  let b: *u8 = ((m as i64) + i * 32) as *u8
  var k: i64 = 0
  while k < 16 { b[k] = addr[k]; k = k + 1 }
  let rp: *i64 = ((m as i64) + i * 32 + 16) as *i64
  rp[0] = ridx
  rp[1] = now
  return i }
234// fragment+send one room-frame to every UDP member of ridx (except member index `except`).
235func sig2_udp_fanout(ufd: i64, m: *u8, ridx: i64, except: i64, dkind: i64, body: *u8, blen: i64,
                   packbuf: *u8, relay_seq_p: *i64) -> i64 {
  let np: i64 = dgm_pack_frame(dkind, "RELAY000" as *u8, relay_seq_p[0], body, blen, packbuf, 131072)
  if np <= 0 { return 0 }
  relay_seq_p[0] = relay_seq_p[0] + 1
  var i: i64 = 0
  while i < NX_SIG2_UDPM {
      let rp: *i64 = ((m as i64) + i * 32 + 16) as *i64
      if rp[0] == ridx { if i != except {
          let dst: *u8 = ((m as i64) + i * 32) as *u8
          var o: i64 = 0
          var p: i64 = 0
          while p < np {
              let pl: i64 = (packbuf[o] & 0xff) + ((packbuf[o+1] & 0xff) << 8)
              sys_sendto(ufd, ((packbuf as i64) + o + 2) as *u8, pl, 0, dst, 16)
              o = o + 2 + pl
              p = p + 1
          }
      } }
      i = i + 1
  }
  return 0 }
257
258func 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 }
259func 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 }
260
261// Emit one telemetry line per ACTIVE room: peers + per-peer relayed/fails.  Cumulative
262// (rising fails on a peer localizes the bottleneck).  Guarded; never crashes the relay.
263func sig2_emit_tel(rooms: *u8, tel: *i64, now: i64) -> i64 {
  let lf: i64 = sys_openat_append(NX_SIG2_TEL_LOG, 420)
  if lf < 0 { return 0 }
  var ri: i64 = 0
  while ri < NX_SIG2_MAX_ROOMS {
      let r: *i64 = sig2_room_ptr(rooms, ri)
      if r[R_USED] == 1 {
          sig2_telw(lf, "ROOMTEL t=" as *u8); sig2_teln(lf, now)
          sig2_telw(lf, " room=" as *u8); sig2_teln(lf, ri)
          sig2_telw(lf, " peers=" as *u8); sig2_teln(lf, r[R_NPEERS])
          var p: i64 = 0
          while p < NX_SIG2_MAX_PEERS {
              if r[R_FD0 + p] != 0 - 1 {
                  let ti: i64 = (ri * NX_SIG2_MAX_PEERS + p) * 2
                  sig2_telw(lf, " p" as *u8); sig2_teln(lf, p)
                  sig2_telw(lf, ":relayed=" as *u8); sig2_teln(lf, tel[ti])
                  sig2_telw(lf, ",fails=" as *u8); sig2_teln(lf, tel[ti + 1])
              }
              p = p + 1
          }
          sig2_telw(lf, "\n" as *u8)
      }
      ri = ri + 1
  }
  sys_close(lf)
  return 0
289}
290
291// conn slot of an fd. -1 = not a WSS conn (e.g. UDP-bridge origin).
292func sig2_slot_of_fd(conns: *u8, fd: i64) -> i64 {
  if fd < 0 { return 0 - 1 }
  var i: i64 = 0
  while i < NX_SIG2_MAX_CONNS {
      let c: *i64 = sig2_conn_ptr(conns, i)
      if c[0] == fd { return i }
      i = i + 1
  }
  return 0 - 1 }
301
302// SFU pairs are keyed by MEMBER id (the 8B id every binary frame carries), NOT conn slot: the TIER-2
303// client hops sticky lanes (separate conns, switch at a key) and reconnects -- the pair's out_seq must
304// stay MONOTONIC across those or the receiver's dedupe would drop everything after a lane switch.
305// find-or-claim; a wrap-claim recycles the oldest slot and clears its pairs (organ reset = safe reuse).
306func sig2_memidx(sfu_mem: *i64, mem_ids: *u8, mcount_p: *i64, id: *u8) -> i64 {
  var i: i64 = 0
  var n: i64 = mcount_p[0]
  if n > SFU_MAXC { n = SFU_MAXC }
  while i < n {
      let e: *u8 = ((mem_ids as i64) + i * 8) as *u8
      var eq: i64 = 1
      var q: i64 = 0
      while q < 8 { if e[q] != id[q] { eq = 0; q = 8 } else { q = q + 1 } }
      if eq == 1 { return i }
      i = i + 1
  }
  let idx: i64 = mcount_p[0] % SFU_MAXC
  mcount_p[0] = mcount_p[0] + 1
  if mcount_p[0] > SFU_MAXC { sfu_reset_conn(sfu_mem, idx) }   // recycled slot -> stale pairs cleared
  let d: *u8 = ((mem_ids as i64) + idx * 8) as *u8
  var w: i64 = 0
  while w < 8 { d[w] = id[w]; w = w + 1 }
  return idx }
325
326// Broadcast a data frame to every OTHER peer in the room; counts per-peer relayed/fails into tel[].
327// SFU (task #41): 0x57 video frames get a per-receiver verdict (nx_sfu_select: which LAYER this receiver
328// takes, switch only at target-layer key) + a rewritten per-pair seq so the forwarded subset stays
329// dedupe/P-chain-legal. Text/audio/chat/FEC broadcast as before; UDP-bridged frames (from_fd=-1 -> no
330// sender slot) and default-HI pairs behave EXACTLY like the old relay by construction.
331// room candidate list for activity ranking: member idxs of every conn in room ridx (deduped by memidx).
332// Returns count. cand cap = NX_SIG2_MAX_CONNS.
333func sig2_room_cand(conns: *u8, cmx: *i64, ridx: i64, cand: *i64) -> i64 {
  var n: i64 = 0
  var i: i64 = 0
  while i < NX_SIG2_MAX_CONNS {
      let c: *i64 = sig2_conn_ptr(conns, i)
      if c[0] >= 0 { if c[1] == ridx { if cmx[i] >= 0 {
          var dup: i64 = 0
          var q: i64 = 0
          while q < n { if cand[q] == cmx[i] { dup = 1; q = n } else { q = q + 1 } }
          if dup == 0 { cand[n] = cmx[i]; n = n + 1 }
      } } }
      i = i + 1
  }
  return n }
347
348// sfuctx: [0]=sfu_mem [1]=conns [2]=conn_ids [3]=mem_ids [4]=mcount_p [5]=conn_memidx(*i64)
349//         [6]=dom state (*i64: per room [dom_midx, last_check_us] x NX_SIG2_MAX_ROOMS) [7]=cand scratch.
350// Sender identity = the frame's OWN id bytes [1..8] (works for any origin, lane-stable). Receiver identity
351// = the memidx learned for that conn; a receiver with no learned id yet gets the frame PLAIN (it cannot
352// have subscribed, and plain = default-HI behavior with the sender's original seq -- keys resync later).
353func sig2_broadcast(rooms: *u8, ridx: i64, from_fd: i64,
                  fin: i64, opcode: i64, payload: *u8, payload_len: i64, tel: *i64, sfuctx: *i64) -> i64 {
  let r: *i64 = sig2_room_ptr(rooms, ridx)
  var is_vid: i64 = 0
  var vlayer: i64 = 0
  var vkey: i64 = 0
  var smidx: i64 = 0 - 1
  if opcode == 2 {
      let vinf: *i64 = (sfuctx[9]) as *i64
      if sfu_frame_bits(payload, payload_len, vinf) == 0 {
          smidx = sig2_memidx((sfuctx[0]) as *i64, (sfuctx[3]) as *u8, (sfuctx[4]) as *i64,
                              ((payload as i64) + 1) as *u8)
          if smidx >= 0 { is_vid = 1; vlayer = vinf[0]; vkey = vinf[1]
              sfu_video_seen((sfuctx[0]) as *i64, smidx, sys_now_us())     // rank tie-break freshness
              sig2_lprod_eval(sfuctx, smidx) }   // R5 self-correcting: leavers/joins re-verdict on the sender's own frames
      }
  }
  let cmx: *i64 = (sfuctx[5]) as *i64
  let cand: *i64 = (sfuctx[7]) as *i64
  var ncand: i64 = 0
  if is_vid == 1 { ncand = sig2_room_cand((sfuctx[1]) as *u8, cmx, ridx, cand) }
  var p: i64 = 0
  while p < NX_SIG2_MAX_PEERS {
      let pfd: i64 = r[R_FD0 + p]
      if pfd != 0 - 1 {
          if pfd != from_fd {
              var send: i64 = 1
              let rcs: i64 = sig2_slot_of_fd((sfuctx[1]) as *u8, pfd)
              if is_vid == 1 {
                  if rcs >= 0 {
                      let rmidx: i64 = cmx[rcs]
                      if rmidx >= 0 {
                          // LAST-N first (R4): outside the receiver's top-N active senders -> not sent,
                          // and the pair's layer/seq state stays untouched (re-entry resyncs at a key).
                          if sfu_video_rank_ok((sfuctx[0]) as *i64, smidx, rmidx, cand, ncand, sys_now_us()) == 0 { send = 0 }
                          else {
                              let oseq: i64 = sfu_on_frame((sfuctx[0]) as *i64, smidx, rmidx, vlayer, vkey)
                              if oseq < 0 { send = 0 } else { sfu_rewrite_seq(payload, oseq) }
                          }
                      }
                  }
              }
              if send == 1 {
                  // frame-atomic non-blocking send (nx_sig2_txq): a slow receiver can no longer stall
                  // the room, and a torn frame can no longer poison its stream. Drops are whole-frame
                  // (seq gap -> the client's kreq recovers) and counted in tel like the old send-fails.
                  var rc: i64 = 0
                  if rcs >= 0 { rc = s2tx_send((sfuctx[8]) as *i64, rcs, pfd, fin * 256 + opcode, payload, payload_len) }
                  else { if nx_ws_send_frame_to_fd(pfd, fin, opcode, payload, payload_len) == NX_WSS_OK { rc = 1 } }
                  let ti: i64 = (ridx * NX_SIG2_MAX_PEERS + p) * 2
                  tel[ti] = tel[ti] + 1
                  if rc != 1 { tel[ti + 1] = tel[ti + 1] + 1 }
              }
          }
      }
      p = p + 1
  }
  return 0
411}
412
413func sig2_room_id_off(path: *u8, path_len: i64) -> i64 {
  let prefix: *u8 = "/signal/" as *u8
  if path_len <= 8 { return 0 - 1 }
  var i: i64 = 0
  while i < 8 {
      if path[i] != prefix[i] { return 0 - 1 }
      i = i + 1
  }
  return 8
422}
423
424// handle one inbound datagram. hctx: [0]=rooms [1]=id_pool [2]=id_pool_off_p [3]=members [4]=dgm_rx_mem
425// [5]=frame_out(64KB) [6]=rinfo(12 i64) [7]=packbuf(128KB) [8]=relay_seq_p [9]=now_us. Unjoined senders
426// are dropped (rule 12); a JOIN datagram (payload = room id) enrolls + ACKs; media raw-fans to UDP peers
427// and, when a frame completes reassembly, bridges as ONE WS frame to every WSS member of the room.
428// enroll a UDP sender as a room member; is_casc=1 marks it a CASCADE peer (another relay). Regular joins
429// get an ACK; cascade joins don't (relays don't wait, and the ACK would leak onto the mesh).
430func sig2_udp_enroll(ufd: i64, pkt: *u8, rinfo: *i64, from: *u8, hctx: *i64, cflags: *i64, is_casc: i64) -> i64 {
  let plen: i64 = rinfo[5]
  if plen < 1 { return 0 }
  if plen > 24 { return 0 }
  let rooms: *u8 = (hctx[0]) as *u8
  let idp: *u8 = (hctx[1]) as *u8
  let offp: *i64 = (hctx[2]) as *i64
  let idptr: *u8 = ((pkt as i64) + rinfo[4]) as *u8
  var ridx: i64 = sig2_find_room(rooms, idp, idptr, plen)
  if ridx < 0 { ridx = sig2_alloc_room(rooms, idp, offp, idptr, plen) }
  if ridx < 0 { return 0 }
  let members: *u8 = (hctx[3]) as *u8
  let mi: i64 = sig2_udpm_upsert(members, from, ridx, hctx[9])
  if mi < 0 { return 0 }
  cflags[mi] = is_casc
  if is_casc == 0 {
      let pb: *u8 = (hctx[7]) as *u8
      let rsp: *i64 = (hctx[8]) as *i64
      let np: i64 = dgm_pack_frame(SIG2_DG_ACK, "RELAY000" as *u8, rsp[0], "ok" as *u8, 2, pb, 131072)
      if np == 1 {
          rsp[0] = rsp[0] + 1
          let pl: i64 = (pb[0] & 0xff) + ((pb[1] & 0xff) << 8)
          sys_sendto(ufd, ((pb as i64) + 2) as *u8, pl, 0, from, 16)
      }
  }
  return 0 }
456
457func sig2_udp_handle(ufd: i64, pkt: *u8, n: i64, from: *u8, hctx: *i64) -> i64 {
  let rinfo: *i64 = (hctx[6]) as *i64
  if dgm_parse_pkt(pkt, n, rinfo) != 0 { return 0 }
  let dkind: i64 = rinfo[0]
  let members: *u8 = (hctx[3]) as *u8
  let now: i64 = hctx[9]
  let cflags: *i64 = (hctx[10]) as *i64
  if dkind == SIG2_DG_JOIN { return sig2_udp_enroll(ufd, pkt, rinfo, from, hctx, cflags, 0) }
  if dkind == SIG2_DG_CASCADE { return sig2_udp_enroll(ufd, pkt, rinfo, from, hctx, cflags, 1) }
  let mi: i64 = sig2_udpm_find(members, from)
  if mi < 0 { return 0 }
  let rp: *i64 = ((members as i64) + mi * 32 + 16) as *i64
  let ridx: i64 = rp[0]
  rp[1] = now
  let from_cascade: i64 = cflags[mi]
  // RAW passthrough to the other UDP members. SPLIT-HORIZON: a frame from a cascade peer goes to LOCAL
  // members only (never back to another cascade relay) -> it crosses each inter-relay link exactly once.
  var i: i64 = 0
  while i < NX_SIG2_UDPM {
      let orp: *i64 = ((members as i64) + i * 32 + 16) as *i64
      if orp[0] == ridx { if i != mi {
          var send_it: i64 = 1
          if from_cascade == 1 { if cflags[i] == 1 { send_it = 0 } }
          if send_it == 1 { sys_sendto(ufd, pkt, n, 0, ((members as i64) + i * 32) as *u8, 16) }
      } }
      i = i + 1
  }
  // BRIDGE to the WSS plane: a complete dgram frame -> one WS frame to every WSS member
  let fout: *u8 = (hctx[5]) as *u8
  let r: i64 = dgm_rx_add((hctx[4]) as *u8, pkt, n, fout, 65536, rinfo)
  if r > 0 {
      var op: i64 = 0
      if rinfo[0] == SIG2_DG_ROOMFRAME { op = 2 }
      if rinfo[0] == SIG2_DG_TEXT { op = 1 }
      if op != 0 {
          let rooms2: *u8 = (hctx[0]) as *u8
          let rr: *i64 = sig2_room_ptr(rooms2, ridx)
          var p2: i64 = 0
          while p2 < NX_SIG2_MAX_PEERS {
              let pfd: i64 = rr[R_FD0 + p2]
              if pfd != 0 - 1 {
                  // frame-atomic non-blocking (hctx[11]=conns, [12]=txctx); a UDP burst can no
                  // longer stall the WSS plane on one slow browser
                  let bci: i64 = sig2_slot_of_fd((hctx[11]) as *u8, pfd)
                  if bci >= 0 { s2tx_send((hctx[12]) as *i64, bci, pfd, 256 + op, fout, r) }
                  else { nx_ws_send_frame_to_fd(pfd, 1, op, fout, r) }
              }
              p2 = p2 + 1
          }
      }
  }
  return 0 }
509
510// QOE TEE (extends the ROOMTEL sense layer, same log): a client's periodic {"type":"qoe",...} beacon
511// (fps_tx / worst fps_rx / rtt / backpressure skips / lanes up / roster-vs-tiles mismatch) is appended RAW
512// the moment it transits the relay. Fire-and-forget (Cardinal 14): a tee failure never disturbs the relay.
513// nx_health_eval aggregates -> health.json + ledger trend, so "my son had 4fps" is VISIBLE on the
514// dashboard without anyone having to report it (operator 2026-07-05: the field call had NO telemetry).
515func sig2_tee_qoe(pp: *u8, plen: i64, ridx: i64) -> i64 {
  if plen < 12 { return 0 }
  if plen > 512 { return 0 }                    // beacons are tiny; never log media-sized text
  let ndl: *u8 = "\"type\":\"qoe\"" as *u8      // 12 bytes
  var found: i64 = 0
  var i: i64 = 0
  while i + 12 <= plen {
      var j: i64 = 0
      var ok: i64 = 1
      while j < 12 { if pp[i + j] != ndl[j] { ok = 0; j = 12 } else { j = j + 1 } }
      if ok == 1 { found = 1; i = plen }
      i = i + 1
  }
  if found == 0 { return 0 }
  let lf: i64 = sys_openat_append(NX_SIG2_TEL_LOG, 420)
  if lf < 0 { return 0 }
  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)
  sig2_telw(lf, " room=" as *u8); sig2_teln(lf, ridx)
  sig2_telw(lf, " " as *u8)
  sys_write(lf, pp, plen)
  sig2_telw(lf, "\n" as *u8)
  sys_close(lf)
  return 0
538}
539
540// DOMINANT-SPEAKER plane (task #41 R4). Speech scored by LPC-frame BYTE COUNT (LPC compresses silence to
541// ~nothing, speech keeps high-entropy residuals -> payload length IS an energy proxy; zero decode, zero
542// client trust). Every ~500ms per room: argmax + hysteresis (organ); on change, announce kind 0x5A
543// [dom_id 8B] to the room -- clients highlight that tile; old clients ignore unknown kinds.
544const SIG2_DOM_CHECK_US: i64 = 500000
545func sig2_audio_activity(fd: i64, ridx: i64, pp: *u8, plen: i64, sfuctx: *i64, midx: i64) -> i64 {
  if midx < 0 { return 0 }
  let now: i64 = sys_now_us()
  sfu_audio_bytes((sfuctx[0]) as *i64, midx, plen, now)
  let dom: *i64 = (sfuctx[6]) as *i64
  let d: i64 = ridx * 2
  if now - dom[d + 1] < SIG2_DOM_CHECK_US { return 0 }
  dom[d + 1] = now
  let cmx: *i64 = (sfuctx[5]) as *i64
  let cand: *i64 = (sfuctx[7]) as *i64
  let nc: i64 = sig2_room_cand((sfuctx[1]) as *u8, cmx, ridx, cand)
  let nd: i64 = sfu_dominant((sfuctx[0]) as *i64, cand, nc, dom[d], now)
  if nd == dom[d] { return 0 }
  if nd < 0 { return 0 }
  dom[d] = nd
  // announce: [0x5A][dom_id 8B][seq4=0][pad] = 16B binary room frame, broadcast to EVERY member
  let ann: *u8 = ((sfuctx[9]) + 64) as *u8
  ann[0] = 0x5A as u8
  let mid: *u8 = ((sfuctx[3]) + nd * 8) as *u8
  var i: i64 = 0
  while i < 8 { ann[1 + i] = mid[i]; i = i + 1 }
  i = 9
  while i < 16 { ann[i] = 0 as u8; i = i + 1 }
  sig2_broadcast_plain(sfuctx, ridx, ann, 16)
  return 0 }
570// plain announce to every conn in room (no SFU, no telemetry counters -- control chatter);
571// frame-atomic non-blocking via the conn's send queue (nx_sig2_txq).
572func sig2_broadcast_plain(sfuctx: *i64, ridx: i64, buf: *u8, n: i64) -> i64 {
  let conns: *u8 = (sfuctx[1]) as *u8
  var i: i64 = 0
  while i < NX_SIG2_MAX_CONNS {
      let c: *i64 = sig2_conn_ptr(conns, i)
      if c[0] >= 0 { if c[1] == ridx { s2tx_send((sfuctx[8]) as *i64, i, c[0], 258, buf, n) } }
      i = i + 1
  }
  return 0 }
581
582// LAYER SUSPENSION (task #41 R5, dynacast-class): if the LO-needed verdict for sender member smidx
583// CHANGED, tell that sender: LPROD 0x5B [id8="RELAYCTL"][seq4=0][want:1] on every conn carrying smidx.
584// Senders then produce LO only while somebody actually consumes it (battery + uplink saved).
585func sig2_lprod_eval(sfuctx: *i64, smidx: i64) -> i64 {
  if smidx < 0 { return 0 }
  if sfu_lprod_check((sfuctx[0]) as *i64, smidx) == 0 { return 0 }
  let want: i64 = sfu_lprod_last((sfuctx[0]) as *i64, smidx)
  let msg: *u8 = ((sfuctx[9]) + 128) as *u8
  msg[0] = 0x5B as u8
  let rid: *u8 = "RELAYCTL" as *u8
  var i: i64 = 0
  while i < 8 { msg[1 + i] = rid[i]; i = i + 1 }
  i = 9
  while i < 13 { msg[i] = 0 as u8; i = i + 1 }
  msg[13] = want as u8
  let conns: *u8 = (sfuctx[1]) as *u8
  let cmx: *i64 = (sfuctx[5]) as *i64
  var s: i64 = 0
  while s < NX_SIG2_MAX_CONNS {
      let c: *i64 = sig2_conn_ptr(conns, s)
      if c[0] >= 0 { if cmx[s] == smidx { s2tx_send((sfuctx[8]) as *i64, s, c[0], 258, msg, 14) } }
      s = s + 1
  }
  return 1 }
606
607// Read one frame from fd; relay/control.  0 = keep, 1 = drop conn.  wuctx (R2 bridge): [0]=udp_fd
608// [1]=udp_members [2]=packbuf [3]=relay_seq_p -- every WSS frame ALSO fans out to the room's UDP members
609// (text -> SIG2_DG_TEXT, binary -> SIG2_DG_ROOMFRAME); wuctx[0]<0 disables (plane not up).
610func sig2_handle_frame(fd: i64, ridx: i64, rooms: *u8, buf: *u8, buf_cap: i64, tel: *i64, wuctx: *i64, sfuctx: *i64) -> i64 {
  let f_raw: *u8 = ((sfuctx[9]) + 192) as *u8
  let f: *WsFrame = f_raw as *WsFrame
  let v: i64 = nx_ws_read_frame_from_fd(fd, buf, buf_cap, f)
  if v != NX_WSS_OK { return 1 }
  if f.opcode == WS_OP_CLOSE { return 1 }
  if f.opcode == WS_OP_PING {
      let pp: *u8 = (buf as i64 + f.payload_off) as *u8
      nx_ws_send_pong(fd, pp, f.payload_len)
      return 0
  }
  if f.opcode == WS_OP_PONG { return 0 }
  let pp: *u8 = (buf as i64 + f.payload_off) as *u8
  if f.opcode == WS_OP_TEXT { sig2_tee_qoe(pp, f.payload_len, ridx) }   // sense layer; relay unchanged
  if f.opcode == 2 { if f.payload_len >= 13 {
      let mslot: i64 = sig2_slot_of_fd((sfuctx[1]) as *u8, fd)
      if mslot >= 0 {
          // learn this conn's member id (every binary room frame carries the sender's own id at [1..8])
          // + resolve its MEMBER idx (the lane-stable SFU key) into the per-conn cache
          let cid: *u8 = ((sfuctx[2]) + mslot * 8) as *u8
          var li: i64 = 0
          while li < 8 { cid[li] = pp[1 + li]; li = li + 1 }
          let cmx: *i64 = (sfuctx[5]) as *i64
          cmx[mslot] = sig2_memidx((sfuctx[0]) as *i64, (sfuctx[3]) as *u8, (sfuctx[4]) as *i64, cid)
          let fkind: i64 = pp[0] & 0xff
          if fkind == 0x4C { sig2_audio_activity(fd, ridx, pp, f.payload_len, sfuctx, cmx[mslot]) }
          if fkind == 0x41 { sig2_audio_activity(fd, ridx, pp, f.payload_len, sfuctx, cmx[mslot]) }
          if fkind == 0x6C { sig2_audio_activity(fd, ridx, pp, f.payload_len, sfuctx, cmx[mslot]) }   // E2EE LPC: length-based
          if fkind == 0x61 { sig2_audio_activity(fd, ridx, pp, f.payload_len, sfuctx, cmx[mslot]) }   // E2EE PCM: energy still works
          if fkind == 0x52 {
              // RECEIVER REPORT (TWCC-class, ladder feedback): [target8 at 13..20][rxfps:1 at 21]...
              // routed to the TARGET SENDER's conns ONLY (never broadcast) -- the sender's ladder learns
              // what each receiver ACTUALLY gets (sender-blindness was the last congestion gap).
              if f.payload_len >= 22 {
                  let conns3: *u8 = (sfuctx[1]) as *u8
                  var rr_i: i64 = 0
                  while rr_i < NX_SIG2_MAX_CONNS {
                      let c3: *i64 = sig2_conn_ptr(conns3, rr_i)
                      if c3[0] >= 0 { if c3[1] == ridx { if rr_i != mslot {
                          let tcid3: *u8 = ((sfuctx[2]) + rr_i * 8) as *u8
                          var eq3: i64 = 1
                          var q3: i64 = 0
                          while q3 < 8 { if tcid3[q3] != pp[13 + q3] { eq3 = 0; q3 = 8 } else { q3 = q3 + 1 } }
                          if eq3 == 1 { s2tx_send((sfuctx[8]) as *i64, rr_i, c3[0], 258, pp, f.payload_len) }
                      } } }
                      rr_i = rr_i + 1
                  }
              }
              return 0
          }
          if fkind == 0x59 {
              // LNSET (task #41 R4): receiver's last-N cap [n:1], device-derived by the client
              // (measured decode capability), NOT a server-hardcoded number. Consumed, never broadcast.
              if f.payload_len >= 14 { sfu_set_lastn((sfuctx[0]) as *i64, cmx[mslot], pp[13] & 0xff) }
              return 0
          }
          if fkind == 0x58 {
              // LSUB (task #41): receiver->relay layer subscription. CONSUMED here -- never broadcast,
              // never bridged. payload = [target_id 8B][want 1B]; target must be a member of THIS room
              // (some conn in the room has learned that id).
              let tid: *u8 = ((sfuctx[9]) + 384) as *u8
              let want: i64 = sfu_parse_lsub(pp, f.payload_len, tid)
              if want >= 0 {
                  let conns2: *u8 = (sfuctx[1]) as *u8
                  var ts: i64 = 0
                  while ts < NX_SIG2_MAX_CONNS {
                      let c2: *i64 = sig2_conn_ptr(conns2, ts)
                      if c2[0] >= 0 { if c2[1] == ridx { if ts != mslot {
                          let tcid: *u8 = ((sfuctx[2]) + ts * 8) as *u8
                          var eq: i64 = 1
                          var q: i64 = 0
                          while q < 8 { if tcid[q] != tid[q] { eq = 0; q = 8 } else { q = q + 1 } }
                          if eq == 1 {
                              let tmidx: i64 = sig2_memidx((sfuctx[0]) as *i64, (sfuctx[3]) as *u8, (sfuctx[4]) as *i64, tid)
                              sfu_want((sfuctx[0]) as *i64, tmidx, cmx[mslot], want)
                              sig2_lprod_eval(sfuctx, tmidx)        // R5: tell the target sender if LO-need flipped
                              ts = NX_SIG2_MAX_CONNS
                          }
                      } } }
                      if ts < NX_SIG2_MAX_CONNS { ts = ts + 1 }
                  }
              }
              return 0
          }
      }
  } }
  sig2_broadcast(rooms, ridx, fd, f.fin, f.opcode, pp, f.payload_len, tel, sfuctx)
  if wuctx[0] >= 0 {
      var dk: i64 = 0
      if f.opcode == WS_OP_TEXT { dk = SIG2_DG_TEXT }
      if f.opcode == 2 { dk = SIG2_DG_ROOMFRAME }
      if dk != 0 { sig2_udp_fanout(wuctx[0], (wuctx[1]) as *u8, ridx, 0 - 1, dk, pp, f.payload_len,
                                   (wuctx[2]) as *u8, (wuctx[3]) as *i64) }
  }
  return 0
705}
706
707// EDGE side of the cascade: announce CASCADE-JOIN(room) to HOME + register HOME as a cascade member here
708// (so HOME's frames split-horizon on this relay too). Called at startup + periodically (keepalive).
709func sig2_cascade_keepalive(ufd: i64, home_addr: *u8, room: *u8, roomlen: i64, hctx: *i64, cflags: *i64) -> i64 {
  let pb: *u8 = (hctx[7]) as *u8
  let rsp: *i64 = (hctx[8]) as *i64
  let np: i64 = dgm_pack_frame(SIG2_DG_CASCADE, "RELAYEDG" as *u8, rsp[0], room, roomlen, pb, 131072)
  if np == 1 {
      rsp[0] = rsp[0] + 1
      let pl: i64 = (pb[0] & 0xff) + ((pb[1] & 0xff) << 8)
      sys_sendto(ufd, ((pb as i64) + 2) as *u8, pl, 0, home_addr, 16)
  }
  let rooms: *u8 = (hctx[0]) as *u8
  let idp: *u8 = (hctx[1]) as *u8
  let offp: *i64 = (hctx[2]) as *i64
  var ridx: i64 = sig2_find_room(rooms, idp, room, roomlen)
  if ridx < 0 { ridx = sig2_alloc_room(rooms, idp, offp, room, roomlen) }
  if ridx >= 0 {
      let members: *u8 = (hctx[3]) as *u8
      let mi: i64 = sig2_udpm_upsert(members, home_addr, ridx, hctx[9])
      if mi >= 0 { cflags[mi] = 1 }
  }
  return 0 }
729
730func main(argc: i64, argv: *i64) -> i64 {
  // THE 2026-07-05 OUTAGE FIX (supervisor reap: sig=13 crash-loop): a phone dropping mid-relay left a
  // dead socket; the next broadcast write raised SIGPIPE and KILLED THE WHOLE DAEMON -- every lane in
  // every room dropped ("relay: reconnecting..."), chronically, for every call. SIG_IGN makes that
  // write return -EPIPE instead; sig2_broadcast already counts it (tel send_fails) and the dead peer's
  // read path closes it. One dying phone must never take down the room again -- by construction.
  nx_signal_ignore(NX_SIGPIPE)
  // ---- CASCADE / multi-instance args (Octo-class). Default (no args) = today's single relay EXACTLY.
  // argv[3]=wss_port argv[4]=udp_port argv[5]=cascade_home_ip argv[6]=cascade_home_port argv[7]=cascade_room
  var wss_port: i64 = NX_SIG2_PORT
  var udp_port: i64 = NX_SIG2_UDP_PORT
  if argc > 3 { let v: i64 = sig2_atoi((argv[3]) as *u8); if v > 0 { wss_port = v } }
  if argc > 4 { let v: i64 = sig2_atoi((argv[4]) as *u8); if v > 0 { udp_port = v } }
  var casc_on: i64 = 0
  let casc_home: *u8 = sys_mmap(16)
  let casc_room: *u8 = sys_mmap(32)
  var casc_roomlen: i64 = 0
  if argc > 7 {
      casc_home[0] = 2 as u8; casc_home[1] = 0 as u8
      let hp: i64 = sig2_atoi((argv[6]) as *u8)
      casc_home[2] = ((hp >> 8) & 0xff) as u8; casc_home[3] = (hp & 0xff) as u8
      sig2_parse_ip((argv[5]) as *u8, ((casc_home as i64) + 4) as *u8)
      var cz: i64 = 8; while cz < 16 { casc_home[cz] = 0 as u8; cz = cz + 1 }
      let rr: *u8 = (argv[7]) as *u8
      var rl: i64 = 0; while rr[rl] != (0 as u8) { casc_room[rl] = rr[rl]; rl = rl + 1 }
      casc_roomlen = rl
      casc_on = 1
  }
  let sfd: i64 = sys_socket(AF_INET, SOCK_STREAM, 0)
  if sfd < 0 { sys_exit(101); return 101 }
  let opt: *u8 = sys_mmap(8)
  opt[0] = 1 as u8; opt[1] = 0 as u8; opt[2] = 0 as u8; opt[3] = 0 as u8
  sys_setsockopt(sfd, SOL_SOCKET, SO_REUSEADDR, opt, 4)
  let addr: *u8 = sys_mmap(16)
  sig2_sockaddr_loopback(addr, wss_port)
  if sys_bind(sfd, addr, 16) < 0 { sys_exit(102); return 102 }
  if sys_listen(sfd, NX_SIG2_ACCEPT_BACKLOG) < 0 { sys_exit(103); return 103 }
  sys_write(2, "nx_signaling_v2 on 127.0.0.1:8445 (N-party rooms, RFC 6455)\n" as *u8, 61)
768
  let rooms: *u8 = sys_mmap(NX_SIG2_MAX_ROOMS * NX_SIG2_ROOM_BYTES)
  let id_pool: *u8 = sys_mmap(NX_SIG2_ROOM_ID_POOL_BYTES)
  let id_pool_off_p: *i64 = sys_mmap(16) as *i64
  id_pool_off_p[0] = 0
773
  let conns: *u8 = sys_mmap(NX_SIG2_MAX_CONNS * NX_SIG2_CONN_BYTES)
  var i: i64 = 0
  while i < NX_SIG2_MAX_CONNS {
      let c: *i64 = sig2_conn_ptr(conns, i)
      c[0] = 0 - 1   // fd
      c[1] = 0 - 1   // room_idx
      i = i + 1
  }
782
  // ---- R2: the SOVEREIGN UDP plane socket (:8471, INADDR_ANY -- native clients hit it directly) ----
  let ufd: i64 = sys_socket(AF_INET, SOCK_DGRAM, 0)
  var udp_up: i64 = 0 - 1
  let uaddr0: *u8 = sys_mmap(16)
  uaddr0[0] = 2 as u8; uaddr0[1] = 0 as u8
  uaddr0[2] = ((udp_port >> 8) & 0xff) as u8
  uaddr0[3] = (udp_port & 0xff) as u8
  var uz: i64 = 4
  while uz < 16 { uaddr0[uz] = 0 as u8; uz = uz + 1 }
  if ufd >= 0 { if sys_bind(ufd, uaddr0, 16) >= 0 { udp_up = ufd
      sys_write(2, "sig2: UDP plane on 0.0.0.0:8471 (sovereign dgram)\n" as *u8, 50) } }
  let udpm: *u8 = sys_mmap(NX_SIG2_UDPM * 32)
  var um: i64 = 0
  while um < NX_SIG2_UDPM { let urp: *i64 = ((udpm as i64) + um * 32 + 16) as *i64; urp[0] = 0 - 1; um = um + 1 }
  let dgmem: *u8 = sys_mmap(DGM_RX_REGION)
  dgm_rx_init(dgmem)
  let ufout: *u8 = sys_mmap(65536)
  let urinfo: *i64 = sys_mmap(8 * 12) as *i64
  let upack: *u8 = sys_mmap(131072)
  let urseq: *i64 = sys_mmap(16) as *i64
  urseq[0] = 1
  let urecv: *u8 = sys_mmap(2048)
  let ufrom: *u8 = sys_mmap(16)
  let ufromlen: *i64 = sys_mmap(16) as *i64
  let cascade_flags: *i64 = sys_mmap(8 * NX_SIG2_UDPM) as *i64   // per UDP member: 1 = a cascade relay peer
  var cf: i64 = 0
  while cf < NX_SIG2_UDPM { cascade_flags[cf] = 0; cf = cf + 1 }
  let hctx: *i64 = sys_mmap(8 * 13) as *i64
  hctx[0] = rooms as i64; hctx[1] = id_pool as i64; hctx[2] = id_pool_off_p as i64
  hctx[3] = udpm as i64; hctx[4] = dgmem as i64; hctx[5] = ufout as i64
  hctx[6] = urinfo as i64; hctx[7] = upack as i64; hctx[8] = urseq as i64; hctx[9] = 0
  hctx[10] = cascade_flags as i64   // [11]=conns [12]=txctx set below once allocated (UDP->WSS bridge sends)
  let wuctx: *i64 = sys_mmap(8 * 4) as *i64
  wuctx[0] = udp_up; wuctx[1] = udpm as i64; wuctx[2] = upack as i64; wuctx[3] = urseq as i64
817
  // ---- SFU (task #41): selective-forwarding state, keyed by MEMBER id (lane/reconnect-stable) ----
  let sfu_mem: *i64 = sys_mmap(SFU_REGION) as *i64
  sfu_init(sfu_mem)
  let conn_ids: *u8 = sys_mmap(NX_SIG2_MAX_CONNS * 8)     // member id (8B) learned per conn slot
  let mem_ids: *u8 = sys_mmap(SFU_MAXC * 8)               // member registry: idx -> id
  let mcount_p: *i64 = sys_mmap(16) as *i64
  mcount_p[0] = 0
  let conn_memidx: *i64 = sys_mmap(8 * NX_SIG2_MAX_CONNS) as *i64
  var cmi: i64 = 0
  while cmi < NX_SIG2_MAX_CONNS { conn_memidx[cmi] = 0 - 1; cmi = cmi + 1 }
  // ---- ROOM TOKENS (capability links): secret + protected list are FILES (data-driven; argv override
  // for the gate). No secret OR empty list -> every room open = exactly today's behavior. ----
  var rk_sec_path: *u8 = "knowledge/room_secret.txt" as *u8
  var rk_lst_path: *u8 = "knowledge/rooms_protected.txt" as *u8
  if argc > 1 { rk_sec_path = (argv[1]) as *u8 }
  if argc > 2 { rk_lst_path = (argv[2]) as *u8 }
  let rk_sec_len_p: *i64 = sys_mmap(16) as *i64
  var rk_sec: *u8 = sys_read_file(rk_sec_path, rk_sec_len_p)
  var rk_sec_len: i64 = 0
  if (rk_sec as i64) != 0 {
      rk_sec_len = rk_sec_len_p[0]
      var trimming: i64 = 1                       // trim trailing CR/LF so editors can't break tokens
      while trimming == 1 {
          if rk_sec_len <= 0 { trimming = 0 }
          else {
              let lc: i64 = rk_sec[rk_sec_len - 1] & 0xff
              if lc == 10 { rk_sec_len = rk_sec_len - 1 }
              else { if lc == 13 { rk_sec_len = rk_sec_len - 1 } else { trimming = 0 } }
          }
      }
  }
  let rk_lst_len_p: *i64 = sys_mmap(16) as *i64
  var rk_lst: *u8 = sys_read_file(rk_lst_path, rk_lst_len_p)
  var rk_lst_len: i64 = 0
  if (rk_lst as i64) != 0 { rk_lst_len = rk_lst_len_p[0] }
  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) } }
854
  let dom_tab: *i64 = sys_mmap(8 * 2 * NX_SIG2_MAX_ROOMS) as *i64   // per room: [dom_midx, last_check_us]
  var dmi: i64 = 0
  while dmi < NX_SIG2_MAX_ROOMS { dom_tab[dmi*2] = 0 - 1; dom_tab[dmi*2+1] = 0; dmi = dmi + 1 }
  let cand_scr: *i64 = sys_mmap(8 * NX_SIG2_MAX_CONNS) as *i64      // room-candidate scratch (single-threaded)
  // frame-atomic TX queues (nx_sig2_txq): one 256KB queue per conn slot + head/len + a header scratch.
  // The slab is mmap-virtual -- pages are touched only when a receiver actually backpressures.
  let txq: *u8 = sys_mmap(NX_SIG2_MAX_CONNS * NX_S2TX_QBYTES)
  let txh: *i64 = sys_mmap(8 * NX_SIG2_MAX_CONNS) as *i64
  let txl: *i64 = sys_mmap(8 * NX_SIG2_MAX_CONNS) as *i64
  let txhscr: *u8 = sys_mmap(16)
  let txctx: *i64 = sys_mmap(8 * 4) as *i64
  txctx[0] = txq as i64; txctx[1] = txh as i64; txctx[2] = txl as i64; txctx[3] = txhscr as i64
  hctx[11] = conns as i64; hctx[12] = txctx as i64   // UDP->WSS bridge sends go frame-atomic too
  let sfuctx: *i64 = sys_mmap(8 * 12) as *i64
  sfuctx[0] = sfu_mem as i64; sfuctx[1] = conns as i64; sfuctx[2] = conn_ids as i64
  sfuctx[3] = mem_ids as i64; sfuctx[4] = mcount_p as i64; sfuctx[5] = conn_memidx as i64
  sfuctx[6] = dom_tab as i64; sfuctx[7] = cand_scr as i64
  sfuctx[8] = txctx as i64   // send-queue ctx: every room-frame send goes through s2tx_send
  // seq1134 LEAK FIX (2026-07-28): the frame/relay hot paths called sys_mmap PER EVENT -- with no
  // munmap in this runtime and page-granular maps, ~40 events/sec leaked into the 2.9GB RSS wedge
  // that silently killed the family call (daemon up + listening + relaying nothing). ONE startup
  // region, fixed offsets reused per event; the poll loop is single-threaded, so reuse is race-free
  // by construction. Layout: +0 vinf(32) +64 ann(32) +128 msg(32) +192 fraw(128) +384 tid(16).
  sfuctx[9] = sys_mmap(512) as i64
879
  let n_pollfds: i64 = NX_SIG2_MAX_CONNS + 2
  let pfds: *u8 = sys_mmap(n_pollfds * NX_POLLFD_BYTES)
  nx_pollfd_set(pfds, 0, sfd, NX_POLLIN)
  var pi: i64 = 1
  while pi < n_pollfds { nx_pollfd_set(pfds, pi, 0 - 1, NX_POLLIN); pi = pi + 1 }
  if udp_up >= 0 { nx_pollfd_set(pfds, NX_SIG2_MAX_CONNS + 1, udp_up, NX_POLLIN) }
886
  let frame_buf: *u8 = sys_mmap(NX_SIG2_FRAME_BUF_BYTES)
  let hdr_buf: *u8 = sys_mmap(NX_SIG2_HDR_BUF_BYTES)
  let path_buf: *u8 = sys_mmap(256)
  let key_buf: *u8 = sys_mmap(128)
  let path_n_p: *i64 = sys_mmap(16) as *i64
  let key_n_p: *i64 = sys_mmap(16) as *i64
893
  let tel: *i64 = sys_mmap(8 * NX_SIG2_TEL_SLOTS) as *i64   // per (room,peer) relayed/fails, zero-init
  var last_emit: i64 = sys_now_us()
  // CASCADE edge: establish the link to HOME immediately (don't wait a full tick)
  if casc_on == 1 { if udp_up >= 0 {
      hctx[9] = sys_now_us()
      sig2_cascade_keepalive(udp_up, casc_home, casc_room, casc_roomlen, hctx, cascade_flags)
      sys_write(2, "sig2: CASCADE edge -> HOME (Octo-class inter-relay link)\n" as *u8, 56)
  } }
902
  var keep: i64 = 1
  while keep == 1 {
      let now0: i64 = sys_now_us()
      if now0 - last_emit > NX_SIG2_EMIT_US {
          sig2_emit_tel(rooms, tel, now0); last_emit = now0
          var sw: i64 = 0                                     // R2: sweep UDP members idle >30s (dead NAT bindings)
          while sw < NX_SIG2_UDPM {
              let srp: *i64 = ((udpm as i64) + sw * 32 + 16) as *i64
              if srp[0] >= 0 { if now0 - srp[1] > 30000000 { srp[0] = 0 - 1 } }
              sw = sw + 1
          }
          if casc_on == 1 { if udp_up >= 0 {   // cascade keepalive: refresh the link (both directions) < the 30s sweep
              hctx[9] = now0
              sig2_cascade_keepalive(udp_up, casc_home, casc_room, casc_roomlen, hctx, cascade_flags)
          } }
      }
      // TX-QUEUE event sweep: a conn with queued bytes also polls POLLOUT so its queue drains the
      // moment the socket has room (frame-atomic continuation -- see nx_sig2_txq).
      var pm: i64 = 0
      while pm < NX_SIG2_MAX_CONNS {
          let cpm: *i64 = sig2_conn_ptr(conns, pm)
          if cpm[0] != 0 - 1 {
              var pev: i64 = NX_POLLIN
              if txl[pm] > 0 { pev = NX_POLLIN | NX_POLLOUT }
              nx_pollfd_set(pfds, pm + 1, cpm[0], pev)
          }
          pm = pm + 1
      }
      // 5s tick, NOT block-forever: nx_poll(-1) lowers to ppoll with a
      // NULL timespec, and that branch miscompiles under the C-bootstrap
      // compiler into a ZERO timespec -> instant return -> 100% CPU spin
      // (found live on the NAS 2026-06-10).  A positive timeout uses the
      // proven path; 0.2 wakeups/sec is negligible.
      let n_ready: i64 = nx_poll(pfds, n_pollfds, 5000)
      if n_ready < 0 { keep = 0 }
      if n_ready == 0 { continue }
939
      if (nx_pollfd_revents(pfds, 0) & NX_POLLIN) != 0 {
          let cfd: i64 = sys_accept(sfd)
          if cfd >= 0 {
              // TCP_NODELAY (IPPROTO_TCP=6, TCP_NODELAY=1): the relay's per-peer sockets carry
              // every broadcast frame back toward the pumps. Nagle here x delayed-ACK on the
              // pump side = the ~40ms burst floor nx_video_qoe_live measured 2026-07-03.
              let snd: *u8 = sys_mmap(4)
              snd[0] = 1 as u8
              snd[1] = 0 as u8
              snd[2] = 0 as u8
              snd[3] = 0 as u8
              sys_setsockopt(cfd, 6, 1, snd, 4)
              let v: i64 = nx_ws_upgrade_handshake(cfd,
                                                   hdr_buf, NX_SIG2_HDR_BUF_BYTES,
                                                   path_buf, 256, path_n_p,
                                                   key_buf, 128, key_n_p)
              if v != NX_WSU_OK {
                  // Cardinal 18: say WHAT failed so a dead join is debuggable.
                  sys_write(2, "sig2: upgrade rejected verdict=" as *u8, 31)
                  let vb: *u8 = sys_mmap(8)
                  vb[0] = (48 + v) as u8
                  vb[1] = 10 as u8
                  sys_write(2, vb, 2)
                  sys_close(cfd)
              } else {
                  let room_off: i64 = sig2_room_id_off(path_buf, path_n_p[0])
                  if room_off < 0 {
                      nx_ws_send_close(cfd, 1008)
                      sys_close(cfd)
                  } else {
                      let id_ptr: *u8 = (path_buf as i64 + room_off) as *u8
                      // ROOM IDENTITY excludes any ?k=... query (a keyed join must land in the SAME room)
                      let id_len: i64 = rk_path_no_query(id_ptr, path_n_p[0] - room_off)
                      // TOKEN WALL: protected base room -> the ?k= token must equal sha256(base||secret).
                      // Empty list / no secret = every room open (today's behavior by construction).
                      var admit: i64 = 1
                      if rk_sec_len > 0 { if rk_lst_len > 0 {
                          let blen: i64 = rk_base_len(id_ptr, id_len)
                          if rk_protected(rk_lst, rk_lst_len, id_ptr, blen) == 1 {
                              admit = 0
                              let kbuf: *u8 = sys_mmap(64)
                              let klen: i64 = rk_parse_k(path_buf, path_n_p[0], kbuf, 32)
                              if klen == RK_TOK_LEN {
                                  let want: *u8 = sys_mmap(32)
                                  rk_token(id_ptr, blen, rk_sec, rk_sec_len, want)
                                  var eqk: i64 = 1
                                  var qk: i64 = 0
                                  while qk < RK_TOK_LEN { if kbuf[qk] != want[qk] { eqk = 0; qk = RK_TOK_LEN } else { qk = qk + 1 } }
                                  if eqk == 1 { admit = 1 }
                              }
                          }
                      } }
                      if admit == 0 {
                          nx_ws_send_close(cfd, 4003)   // keyed room: token absent or wrong
                          sys_close(cfd)
                      } else {
                      var ridx: i64 = sig2_find_room(rooms, id_pool, id_ptr, id_len)
                      if ridx < 0 {
                          ridx = sig2_alloc_room(rooms, id_pool, id_pool_off_p, id_ptr, id_len)
                      }
                      if ridx < 0 {
                          // Cardinal 18: a refused room names WHICH limit refused it (the 1011 class was silent for hours).
                          if ridx == 0 - 2 { sys_write(2, "sig2: room refused: id longer than a slot\n" as *u8, 42) }
                          else { sys_write(2, "sig2: room refused: all room slots in use\n" as *u8, 42) }
                          nx_ws_send_close(cfd, 1011)
                          sys_close(cfd)
                      } else {
                          let slot: i64 = sig2_join(rooms, ridx, cfd)
                          if slot < 0 {
                              nx_ws_send_close(cfd, 1008)   // room full (8 peers)
                              sys_close(cfd)
                          } else {
                              var ci: i64 = 0
                              var placed: i64 = 0
                              while ci < NX_SIG2_MAX_CONNS {
                                  let c: *i64 = sig2_conn_ptr(conns, ci)
                                  if c[0] == 0 - 1 {
                                      if placed == 0 {
                                          c[0] = cfd
                                          c[1] = ridx
                                          txh[ci] = 0            // fresh conn slot -> empty send queue
                                          txl[ci] = 0
                                          nx_pollfd_set(pfds, ci + 1, cfd, NX_POLLIN)
                                          placed = 1
                                      }
                                  }
                                  ci = ci + 1
                              }
                              if placed == 0 {
                                  sig2_leave(rooms, ridx, cfd)
                                  sys_close(cfd)
                              }
                          }
                      }
                      }
                  }
              }
          }
      }
1039
      // R2: drain one datagram per wake (poll re-fires if more are queued -- no nonblock needed)
      if udp_up >= 0 {
          if (nx_pollfd_revents(pfds, NX_SIG2_MAX_CONNS + 1) & NX_POLLIN) != 0 {
              ufromlen[0] = 16
              let rn: i64 = sys_recvfrom(udp_up, urecv, 2048, 0, ufrom, ufromlen)
              if rn > 0 { hctx[9] = now0; sig2_udp_handle(udp_up, urecv, rn, ufrom, hctx) }
          }
      }
1048
      var ci: i64 = 0
      while ci < NX_SIG2_MAX_CONNS {
          let c: *i64 = sig2_conn_ptr(conns, ci)
          if c[0] != 0 - 1 {
              let rev: i64 = nx_pollfd_revents(pfds, ci + 1)
              if rev != 0 {
                  var dropv: i64 = 0
                  if (rev & NX_POLLOUT) != 0 {              // socket has room -> continue the queued bytes
                      if s2tx_drain(txq, txh, txl, ci, c[0]) < 0 { dropv = 1 }
                  }
                  if dropv == 0 {
                      if (rev & (NX_POLLIN | NX_POLLERR | NX_POLLHUP | NX_POLLNVAL)) != 0 {
                          dropv = sig2_handle_frame(c[0], c[1], rooms, frame_buf, NX_SIG2_FRAME_BUF_BYTES, tel, wuctx, sfuctx)
                      }
                  }
                  if dropv == 1 {
                      sig2_leave(rooms, c[1], c[0])
                      sys_close(c[0])
                      c[0] = 0 - 1
                      c[1] = 0 - 1
                      conn_memidx[ci] = 0 - 1              // conn slot free; MEMBER pairs persist (lane
                      txh[ci] = 0                          // hops + reconnects keep seq monotonic)
                      txl[ci] = 0
                      nx_pollfd_set(pfds, ci + 1, 0 - 1, NX_POLLIN)
                  }
              }
          }
          ci = ci + 1
      }
  }
  sys_close(sfd)
  return 0
1081}
nx_signaling_v2.nx source

`nx_signaling_v2.nx` source
