nx_sfu_select.nx source
↩ module page · 241 lines · 11750 B
1// nx_sfu_select.nx -- SOVEREIGN SELECTIVE-FORWARDING core (the jitsi-videobridge value prop, measured our
2// #1 SOTA hole: relay had broadcast=7/selective=0 -> every receiver got every full-rate stream -> multi-
3// party collapse). PURE LOGIC, no sockets: the relay integrates it thinly; this organ is gate-proven alone.
4//
5// MODEL (layers = geometries): a sender may mark each 0x57 frame HI (flags bit1=0, today's clients) or LO
6// (bit1=1, e.g. 160x120 simulcast). Per (sender,receiver) pair the relay forwards EXACTLY ONE layer.
7// A layer SWITCH happens only AT A KEYFRAME of the target layer -- to a geometry-flex receiver the switch
8// is just a legal geometry change at a key: ZERO new client decode logic. The forwarded subset gets its
9// seq REWRITTEN (out_seq++ per forwarded frame, the JVB trick) so dedupe+P-chain rules hold.
10// SAFE BY CONSTRUCTION: default want=HI = exactly today's behavior; want=LO with a sender that never
11// produces LO keeps HI flowing (no starvation); unknown flag bits ignored; senders/receivers cap 64.
12// license_tier: ORIGINAL
13import "nx_syscalls.nx"
14const SFU_MAGIC_65536: i64 = 65536
15const SFU_MAGIC_16777216: i64 = 16777216
16
17const SFU_MAXC: i64 = 128 // relay conn slots (matches NX_SIG2_MAX_CONNS)
18const SFU_PAIRS: i64 = 16384 // 128x128
19// per-pair record: 4 x i64 = [cur_layer(-1 unset), want_layer, out_seq_next, fwd_count]
20const SFU_REC: i64 = 4
21// ACTIVITY plane (dominant-speaker + last-N, task #41 R4). Speech detection is SERVER-side with ZERO
22// decode: the client's audio rides LPC-LOSSLESS frames (0x4C), and LPC compresses silence to almost
23// nothing while speech keeps high-entropy residuals -> the PAYLOAD BYTE COUNT is an honest energy proxy
24// (and unforgeable-cheaply, unlike client-reported levels). Score = bytes in a 2-bucket rolling window.
25const SFU_ACT_BUCKET_US: i64 = 2000000 // bucket width: score covers the last 2..4s of speech
26const SFU_DOM_HYST_N: i64 = 5 // hysteresis 5/4: a challenger needs 1.25x the incumbent's
27const SFU_DOM_HYST_D: i64 = 4 // score to take over -- prevents dominant flapping
28// per-member activity record: 6 x i64 = [bytes_cur, bytes_prev, bucket_epoch, last_video_us, lastn,
29// lprod_last(-1 unannounced)] -- lprod_last = the LO-production verdict last announced to this SENDER
30const SFU_ACTREC: i64 = 6
31const SFU_ACT_OFF: i64 = SFU_REC * SFU_PAIRS // i64 offset where activity plane starts
32const SFU_REGION: i64 = 8 * (SFU_REC * SFU_PAIRS + SFU_ACTREC * SFU_MAXC)
33
34func sfu_init(mem: *i64) -> i64 {
35 var i: i64 = 0
36 while i < SFU_PAIRS * SFU_REC + SFU_ACTREC * SFU_MAXC { mem[i] = 0; i = i + 1 }
37 // cur_layer must start UNSET(-1); want defaults HI(0)
38 var p: i64 = 0
39 while p < SFU_PAIRS { mem[p * SFU_REC] = 0 - 1; p = p + 1 }
40 var m2: i64 = 0
41 while m2 < SFU_MAXC { let a: *i64 = sfu_act_ptr(mem, m2); a[5] = 0 - 1; m2 = m2 + 1 } // lprod unannounced
42 return 0 }
43
44// ---- activity plane ----
45func sfu_act_ptr(mem: *i64, midx: i64) -> *i64 { return ((mem as i64) + 8 * (SFU_ACT_OFF + midx * SFU_ACTREC)) as *i64 }
46// rotate the 2-bucket window to `now`; 2+ buckets idle -> both clear (stale speech never lingers)
47func sfu_act_roll(a: *i64, now_us: i64) -> i64 {
48 let ep: i64 = now_us / SFU_ACT_BUCKET_US
49 if a[2] == ep { return 0 }
50 if a[2] == ep - 1 { a[1] = a[0]; a[0] = 0; a[2] = ep; return 0 }
51 a[0] = 0; a[1] = 0; a[2] = ep
52 return 0 }
53func sfu_audio_bytes(mem: *i64, midx: i64, nbytes: i64, now_us: i64) -> i64 {
54 if midx < 0 { return 0 - 1 }
55 if midx >= SFU_MAXC { return 0 - 1 }
56 if nbytes < 0 { return 0 - 1 }
57 let a: *i64 = sfu_act_ptr(mem, midx)
58 sfu_act_roll(a, now_us)
59 a[0] = a[0] + nbytes
60 return 0 }
61func sfu_act_score(mem: *i64, midx: i64, now_us: i64) -> i64 {
62 if midx < 0 { return 0 }
63 if midx >= SFU_MAXC { return 0 }
64 let a: *i64 = sfu_act_ptr(mem, midx)
65 sfu_act_roll(a, now_us)
66 return a[0] + a[1] }
67func sfu_video_seen(mem: *i64, midx: i64, now_us: i64) -> i64 {
68 if midx < 0 { return 0 - 1 }
69 if midx >= SFU_MAXC { return 0 - 1 }
70 let a: *i64 = sfu_act_ptr(mem, midx)
71 a[3] = now_us
72 return 0 }
73func sfu_set_lastn(mem: *i64, midx: i64, n: i64) -> i64 {
74 if midx < 0 { return 0 - 1 }
75 if midx >= SFU_MAXC { return 0 - 1 }
76 if n < 0 { return 0 - 1 }
77 if n > SFU_MAXC { return 0 - 1 }
78 let a: *i64 = sfu_act_ptr(mem, midx)
79 a[4] = n // 0 = unlimited (default: today's behavior)
80 return 0 }
81// dominant among a CANDIDATE list (the relay passes the room's member idxs): argmax score with 5/4
82// hysteresis vs the incumbent. -1 = nobody scored (silent room keeps the old highlight off).
83func sfu_dominant(mem: *i64, cand: *i64, ncand: i64, incumbent: i64, now_us: i64) -> i64 {
84 var best: i64 = 0 - 1
85 var bs: i64 = 0
86 var i: i64 = 0
87 while i < ncand {
88 let s: i64 = sfu_act_score(mem, cand[i], now_us)
89 if s > bs { bs = s; best = cand[i] }
90 i = i + 1
91 }
92 if best < 0 { return 0 - 1 }
93 if bs == 0 { return 0 - 1 }
94 if incumbent >= 0 { if best != incumbent {
95 let is2: i64 = sfu_act_score(mem, incumbent, now_us)
96 if bs * SFU_DOM_HYST_D < is2 * SFU_DOM_HYST_N { return incumbent } // challenger below 1.25x -> hold
97 } }
98 return best }
99// LAST-N verdict: may sender s's video reach receiver r? r's lastn==0 -> always. Else s must rank
100// inside r's top-N among the candidates by activity score (tie -> most recent video wins).
101func sfu_video_rank_ok(mem: *i64, s: i64, r: i64, cand: *i64, ncand: i64, now_us: i64) -> i64 {
102 if r < 0 { return 1 }
103 if r >= SFU_MAXC { return 1 }
104 let ar: *i64 = sfu_act_ptr(mem, r)
105 let n: i64 = ar[4]
106 if n <= 0 { return 1 }
107 let ss: i64 = sfu_act_score(mem, s, now_us)
108 let asv: *i64 = sfu_act_ptr(mem, s)
109 let sv: i64 = asv[3]
110 var ahead: i64 = 0
111 var i: i64 = 0
112 while i < ncand {
113 let c: i64 = cand[i]
114 if c != s { if c != r {
115 let cs: i64 = sfu_act_score(mem, c, now_us)
116 var beats: i64 = 0
117 if cs > ss { beats = 1 }
118 if cs == ss { let ac: *i64 = sfu_act_ptr(mem, c); if ac[3] > sv { beats = 1 } }
119 ahead = ahead + beats
120 } }
121 i = i + 1
122 }
123 if ahead < n { return 1 }
124 return 0 }
125
126// receiver r subscribes to layer `want` (0=HI 1=LO) for sender s. Bad args = no-op (rule 12).
127func sfu_want(mem: *i64, s: i64, r: i64, want: i64) -> i64 {
128 if s < 0 { return 0 - 1 }
129 if r < 0 { return 0 - 1 }
130 if s >= SFU_MAXC { return 0 - 1 }
131 if r >= SFU_MAXC { return 0 - 1 }
132 if want < 0 { return 0 - 1 }
133 if want > 1 { return 0 - 1 }
134 mem[(s * SFU_MAXC + r) * SFU_REC + 1] = want
135 return 0 }
136
137// a conn slot is reused by a NEW member (join/leave): clear every pair touching idx (as sender AND receiver)
138func sfu_reset_conn(mem: *i64, idx: i64) -> i64 {
139 if idx < 0 { return 0 - 1 }
140 if idx >= SFU_MAXC { return 0 - 1 }
141 var r: i64 = 0
142 while r < SFU_MAXC {
143 let a: i64 = (idx * SFU_MAXC + r) * SFU_REC
144 mem[a] = 0 - 1; mem[a+1] = 0; mem[a+2] = 0; mem[a+3] = 0
145 let b2: i64 = (r * SFU_MAXC + idx) * SFU_REC
146 mem[b2] = 0 - 1; mem[b2+1] = 0; mem[b2+2] = 0; mem[b2+3] = 0
147 r = r + 1
148 }
149 let a: *i64 = sfu_act_ptr(mem, idx) // recycled member: stale speech/lastn must not carry over
150 a[0]=0; a[1]=0; a[2]=0; a[3]=0; a[4]=0; a[5]=0-1
151 return 0 }
152
153// LAYER SUSPENSION (dynacast-class, task #41 R5): does ANY receiver need sender s's LO layer?
154// True if some pair (s,r) WANTS LO or is still FLOWING LO (cur) -- suspension only when truly unwatched.
155func sfu_lo_wanted(mem: *i64, s: i64) -> i64 {
156 if s < 0 { return 0 }
157 if s >= SFU_MAXC { return 0 }
158 var r: i64 = 0
159 while r < SFU_MAXC {
160 let a: i64 = (s * SFU_MAXC + r) * SFU_REC
161 if mem[a+1] == 1 { return 1 }
162 if mem[a] == 1 { return 1 }
163 r = r + 1
164 }
165 return 0 }
166// announce bookkeeping: returns 1 if the verdict CHANGED vs the last announcement (caller then sends
167// LPROD 0x5B to the sender and this records it); 0 = no announcement needed.
168func sfu_lprod_check(mem: *i64, s: i64) -> i64 {
169 if s < 0 { return 0 }
170 if s >= SFU_MAXC { return 0 }
171 let want: i64 = sfu_lo_wanted(mem, s)
172 let a: *i64 = sfu_act_ptr(mem, s)
173 if a[5] == want { return 0 }
174 a[5] = want
175 return 1 }
176func sfu_lprod_last(mem: *i64, s: i64) -> i64 {
177 if s < 0 { return 0 - 1 }
178 if s >= SFU_MAXC { return 0 - 1 }
179 let a: *i64 = sfu_act_ptr(mem, s)
180 return a[5] }
181
182// THE VERDICT. Sender s emitted a video frame (layer, is_key). Should receiver r get it, and with which
183// rewritten seq? Returns -1 = DO NOT forward; else the out_seq (>=0) to write into the frame for r.
184// State machine:
185// cur unset : pair starts on HI semantics (today's behavior) -> treat cur=HI immediately.
186// layer==cur : forward (the flowing layer keeps flowing even if want differs -- no gap until switchable).
187// layer==want and want!=cur and is_key : SWITCH (cur=want) + forward (key resets receiver chain legally).
188// otherwise : drop (target-layer P before its key is undecodable; other-layer traffic isn't ours).
189func sfu_on_frame(mem: *i64, s: i64, r: i64, layer: i64, is_key: i64) -> i64 {
190 if s < 0 { return 0 - 1 }
191 if r < 0 { return 0 - 1 }
192 if s >= SFU_MAXC { return 0 - 1 }
193 if r >= SFU_MAXC { return 0 - 1 }
194 if s == r { return 0 - 1 }
195 var ly: i64 = layer
196 if ly != 0 { ly = 1 } // any nonzero layer bit -> LO (unknown bits don't invent layers)
197 let a: i64 = (s * SFU_MAXC + r) * SFU_REC
198 var cur: i64 = mem[a]
199 let want: i64 = mem[a+1]
200 if cur < 0 { cur = 0; mem[a] = 0 } // first sight -> HI semantics (back-compat by construction)
201 var fwd: i64 = 0
202 if ly == cur { fwd = 1 }
203 if fwd == 0 { if ly == want { if want != cur { if is_key == 1 { mem[a] = want; fwd = 1 } } } }
204 if fwd == 0 { return 0 - 1 }
205 let oseq: i64 = mem[a+2]
206 mem[a+2] = oseq + 1
207 mem[a+3] = mem[a+3] + 1
208 return oseq }
209
210// helper for the relay: parse a 0x57 wire frame's SFU-relevant bits in place.
211// buf = whole room-protocol frame [kind|id8|seq4|payload...]; n = length.
212// info[0]=layer info[1]=is_key ; returns 0 ok, -1 not a video frame / too short.
213func sfu_frame_bits(buf: *u8, n: i64, info: *i64) -> i64 {
214 if n < 14 { return 0 - 1 }
215 let k: i64 = buf[0] & 0xff
216 if k != 0x57 { if k != 0x77 { return 0 - 1 } } // 0x77 = E2EE video: flags still CLEAR at [13], codec encrypted -> full SFU (layer-select + seq-rewrite) stays relay-blind
217 let flags: i64 = buf[13] & 0xff
218 info[1] = flags & 1
219 var ly: i64 = 0
220 if (flags & 2) == 2 { ly = 1 }
221 info[0] = ly
222 return 0 }
223
224// helper for the relay: rewrite the 4-byte LE seq field (offset 9) to out_seq before sending to ONE receiver.
225func sfu_rewrite_seq(buf: *u8, out_seq: i64) -> i64 {
226 buf[9] = (out_seq & 255) as u8
227 buf[10] = ((out_seq / 256) & 255) as u8
228 buf[11] = ((out_seq / SFU_MAGIC_65536) & 255) as u8
229 buf[12] = ((out_seq / SFU_MAGIC_16777216) & 255) as u8
230 return 0 }
231
232// LSUB control frame (kind 0x58, receiver->relay only, NEVER broadcast): payload = [target_id 8B][want 1B].
233// Parses out of the wire frame; info[0..7]=target id bytes... returns want (0/1) or -1 invalid.
234func sfu_parse_lsub(buf: *u8, n: i64, target_id_out: *u8) -> i64 {
235 if n < 22 { return 0 - 1 } // 13 hdr + 8 id + 1 want
236 if (buf[0] & 0xff) != 0x58 { return 0 - 1 }
237 var i: i64 = 0
238 while i < 8 { target_id_out[i] = buf[13 + i]; i = i + 1 }
239 let want: i64 = buf[21] & 0xff
240 if want > 1 { return 0 - 1 }
241 return want }