nx_sig2_txq.nx source

1// nx_sig2_txq.nx -- FRAME-ATOMIC non-blocking WebSocket send queue for the signaling relay. 2// (four-pillar ADDRESS at the transport layer; root cause: the relay's blocking fan-out writes.) 3// 4// THE TWO DEFECTS THIS CLOSES (source-confirmed 2026-07-09, [[project-nishi-videoops-four-pillars]]): 5// 1. BLOCKING WRITES in a single-threaded fan-out: one slow receiver's full socket buffer STALLED 6// every room on the relay (head-of-line). Writes here use sendto(MSG_DONTWAIT) -- the relay 7// NEVER sleeps on a receiver. (Write-side only: the read path keeps its blocking semantics.) 8// 2. PARTIAL-FRAME POISON: _wss_write_n bailed mid-frame on w<=0, leaving a torn WebSocket frame 9// on the stream -- every later byte misparsed (the field's silent video garbage). Here a frame 10// either goes out WHOLE (possibly across POLLOUT continuations from the queue) or is dropped 11// WHOLE before any byte touches the wire. Torn-frame-unavoidable (partial direct write + queue 12// full) is FATAL by design: the caller must treat the conn as poisoned and let the reap path 13// close it -- fail-stop beats fail-dirty. 14// 15// Per-conn state (caller-allocated, indexed by conn slot ci): 16// txq slab: NX_S2TX_QBYTES per conn -- queued raw wire bytes (whole frames / frame remainders) 17// txh[ci]: queue head offset txl[ci]: queued byte count 18// Drop policy: a frame that cannot fit WHOLE in the queue is dropped whole and counted by the 19// caller (video P-gap -> the client's seq check + kreq recovers in ~1 RTT; JSON control re-sends). 20// license_tier: ORIGINAL 21import "nx_syscalls.nx" 22import "nx_websocket_frame.nx" 23 24const NX_S2TX_QBYTES: i64 = 262144 // 256KB/conn ~= 4 worst-case keyframes ~= 250ms drop horizon 25const NX_S2TX_DONTWAIT: i64 = 0x40 // MSG_DONTWAIT: this write NEVER sleeps 26const NX_S2TX_NOSIGNAL: i64 = 0x4000 // MSG_NOSIGNAL: a send to a dead peer returns -EPIPE, never SIGPIPE 27const NX_S2TX_SENDFLAGS: i64 = 0x4040 // DONTWAIT | NOSIGNAL -- self-protecting regardless of signal disposition 28const NX_S2TX_EAGAIN: i64 = 0 - 11 29 30func s2tx_qptr(txq: *u8, ci: i64) -> *u8 { 31 return (txq as i64 + ci * NX_S2TX_QBYTES) as *u8 32} 33 34// non-blocking write of up to n bytes; returns bytes written (>=0), 0 on EAGAIN, -1 fatal. 35// MSG_NOSIGNAL means a write to a closed peer returns -EPIPE (fatal -> caller reaps) and NEVER raises 36// SIGPIPE -- the relay cannot be killed by one dead receiver even if it forgot to ignore the signal. 37func s2tx_wr(fd: i64, p: *u8, n: i64) -> i64 { 38 let w: i64 = sys_sendto(fd, p, n, NX_S2TX_SENDFLAGS, 0 as *u8, 0) 39 if w > 0 { return w } 40 if w == NX_S2TX_EAGAIN { return 0 } 41 return 0 - 1 42} 43 44// Push queued bytes at the socket until empty or EAGAIN. 0 = ok (maybe residue), -1 = fatal. 45func s2tx_drain(txq: *u8, txh: *i64, txl: *i64, ci: i64, fd: i64) -> i64 { 46 let q: *u8 = s2tx_qptr(txq, ci) 47 var run: i64 = 1 48 var rc: i64 = 0 49 while run == 1 { 50 if txl[ci] <= 0 { txh[ci] = 0; txl[ci] = 0; run = 0 } 51 else { 52 let w: i64 = s2tx_wr(fd, (q as i64 + txh[ci]) as *u8, txl[ci]) 53 if w > 0 { txh[ci] = txh[ci] + w; txl[ci] = txl[ci] - w } 54 else { 55 if w == 0 { run = 0 } // EAGAIN: POLLOUT continues later 56 else { rc = 0 - 1; run = 0 } // dead socket: caller reaps the conn 57 } 58 } 59 } 60 return rc 61} 62 63// Send one whole WebSocket frame, frame-atomically, never blocking. 64// tx = [txq, txh, txl, hdr_scratch(>=16B)] finop = fin*256 + opcode 65// Returns 1 = fully sent or queued whole; 0 = dropped whole (queue full, wire untouched); 66// -1 = fatal (socket error, or torn frame unavoidable -> conn must be reaped). 67func s2tx_send(tx: *i64, ci: i64, fd: i64, finop: i64, payload: *u8, plen: i64) -> i64 { 68 let txq: *u8 = (tx[0]) as *u8 69 let txh: *i64 = (tx[1]) as *i64 70 let txl: *i64 = (tx[2]) as *i64 71 let hscr: *u8 = (tx[3]) as *u8 72 let hlen: i64 = ws_build_header(hscr, 16, finop / 256, finop & 0xff, 0, 0, plen) 73 if hlen < 0 { return 0 - 1 } 74 if txl[ci] > 0 { 75 if s2tx_drain(txq, txh, txl, ci, fd) < 0 { return 0 - 1 } 76 } 77 var hoff: i64 = 0 78 var poff: i64 = 0 79 if txl[ci] == 0 { 80 // FAST PATH: queue empty -> write direct as far as the socket takes it, never sleeping 81 var run: i64 = 1 82 while run == 1 { 83 if hoff < hlen { 84 let w: i64 = s2tx_wr(fd, (hscr as i64 + hoff) as *u8, hlen - hoff) 85 if w > 0 { hoff = hoff + w } 86 else { if w == 0 { run = 0 } else { return 0 - 1 } } 87 } else { 88 if poff < plen { 89 let w2: i64 = s2tx_wr(fd, (payload as i64 + poff) as *u8, plen - poff) 90 if w2 > 0 { poff = poff + w2 } 91 else { if w2 == 0 { run = 0 } else { return 0 - 1 } } 92 } else { run = 0 } 93 } 94 } 95 if hoff >= hlen { if poff >= plen { return 1 } } // whole frame on the wire 96 } 97 // Stash the remainder (or the whole frame) -- WHOLE or not at all. 98 let need: i64 = (hlen - hoff) + (plen - poff) 99 let q: *u8 = s2tx_qptr(txq, ci) 100 if txh[ci] > 0 { 101 if txh[ci] + txl[ci] + need > NX_S2TX_QBYTES { // compact once: reclaim the drained head 102 var mi: i64 = 0 103 while mi < txl[ci] { q[mi] = q[txh[ci] + mi]; mi = mi + 1 } 104 txh[ci] = 0 105 } 106 } 107 if txh[ci] + txl[ci] + need > NX_S2TX_QBYTES { 108 if hoff > 0 { return 0 - 1 } // part of THIS frame is already on the wire: 109 if poff > 0 { return 0 - 1 } // dropping now would tear it -> fail-stop 110 return 0 // wire untouched -> drop whole, count it 111 } 112 var wp: i64 = txh[ci] + txl[ci] 113 var k: i64 = hoff 114 while k < hlen { q[wp] = hscr[k]; wp = wp + 1; k = k + 1 } 115 k = poff 116 while k < plen { q[wp] = payload[k]; wp = wp + 1; k = k + 1 } 117 txl[ci] = txl[ci] + need 118 return 1 119} 120 121func main() -> i64 { return 0 }