nx_dgram_media.nx source
↩ module page · 263 lines · 12064 B
1// nx_dgram_media.nx -- R1 of the SOVEREIGN DATAGRAM TRANSPORT (task #15; operator 2026-07-05: "pure nishi
2// sovereign -- no WebTransport; build from the hardware rung up, each rung"). The UDP-class media wire:
3// small MTU-safe packets, fragmentation/reassembly for codec frames (a 7KB keyframe -> ~7 frags), seq
4// dedupe, gap detection (feeds NACK), and the NACK packet format. TCP/WSS gave us ordering+retransmit at
5// the cost of head-of-line blocking (the last structural ceiling vs the UDP rivals); THIS plane gives the
6// opposite trade and lets the ALREADY-GATED organs finally earn their keep: nx_nack (retransmit asks),
7// nx_bwe (rate), nx_jitter_buffer (playout), FEC shards (loss without asks). Runs NATIVE (relay + the
8// native client #22 + NishiOS); 3rd-party browser tabs stay on the WSS compat lane (their sandbox has no
9// raw UDP -- adapter, never architecture). Flat memory + i64 API, caller owns all regions. license_tier: ORIGINAL
10//
11// PACKET (<= DGM_MTU bytes):
12// [magic 'N''D':2][kind:1][sender:8][seq:4 LE][frag_i:1][frag_n:1][flags:1] = 18B header, payload after.
13// kinds: media passthrough (0x57 codec / 0x4C 0x41 audio / 0x43 chat) + 0x4E NACK (payload = [cnt:1][seq:4]xcnt).
14// REASSEMBLY region (caller provides DGM_RX_REGION bytes): 8 sender slots x 4 pending frames x 32 frags.
15
16const DGM_MTU: i64 = 1200
17const DGM_HDR: i64 = 18
18const DGM_FRAG_CAP: i64 = 1182 // DGM_MTU - DGM_HDR
19const DGM_MAX_FRAGS: i64 = 32 // frame cap = 32*1182 = 37824B (a keyframe is ~7KB; 5x headroom)
20const DGM_SENDERS: i64 = 8
21const DGM_PENDING: i64 = 4 // in-flight assemblies per sender
22const DGM_KIND_NACK: i64 = 0x4E
23
24// region layout ---------------------------------------------------------------------------------
25// [0..64) : sender id table (8 x 8B)
26// per sender s (base = 64 + s*DGM_SSTRIDE):
27// +0 hi_seq: i64 : highest seq DELIVERED-or-seen-complete
28// +8 dmask: 4 x i64 : delivered bitmap, 256-seq window (hi_seq-255 .. hi_seq) -- dedupe + gaps.
29// (the gate's first run PROVED 64 was too small: a frame NACK-retransmitted after ~4s of newer
30// traffic fell off the window and was dropped as "ancient" forever; 256 ≈ 17s at 15fps)
31// +40 4 x pending: [seq:i64][frag_n:i64][got_mask:i64][got_cnt:i64][kind:i64][ln 32 x i64][buf 32*1182]
32const DGM_WIN: i64 = 256
33const DGM_PFIX: i64 = 40 // 5 i64 fixed fields per pending
34const DGM_PLENS: i64 = 256 // 32 x i64 per-frag lengths
35const DGM_PBUF: i64 = 37824
36const DGM_PSTRIDE: i64 = 38120 // DGM_PFIX + DGM_PLENS + DGM_PBUF
37const DGM_SSTRIDE: i64 = 152520 // 40 + 4*DGM_PSTRIDE
38const DGM_RX_REGION: i64 = 1220224 // 64 + 8*DGM_SSTRIDE
39
40// 256-bit delivered-mask helpers (mp -> 4 x i64). bit d = "seq (hi_seq - d) was delivered".
41func dgm_mask_test(mp: *i64, d: i64) -> i64 { return (mp[d >> 6] >> (d & 63)) & 1 }
42func dgm_mask_set(mp: *i64, d: i64) -> i64 { mp[d >> 6] = mp[d >> 6] | (1 << (d & 63)); return 0 }
43func dgm_mask_shift(mp: *i64, s: i64) -> i64 { // left-shift the whole 256-bit window by s (new frames)
44 if s >= DGM_WIN { mp[0]=0; mp[1]=0; mp[2]=0; mp[3]=0; return 0 }
45 let ws: i64 = s >> 6
46 let bs: i64 = s & 63
47 var i: i64 = 3
48 while i >= 0 {
49 var v: i64 = 0
50 if i - ws >= 0 { v = mp[i - ws] << bs }
51 if bs > 0 { if i - ws - 1 >= 0 { v = v | ((mp[i - ws - 1] >> (64 - bs)) & ((1 << bs) - 1)) } }
52 mp[i] = v
53 i = i - 1
54 }
55 return 0 }
56
57func dgm_wr_u32(p: *u8, off: i64, v: i64) -> i64 {
58 p[off] = (v & 0xff) as u8
59 p[off+1] = ((v >> 8) & 0xff) as u8
60 p[off+2] = ((v >> 16) & 0xff) as u8
61 p[off+3] = ((v >> 24) & 0xff) as u8
62 return 0 }
63func dgm_rd_u32(p: *u8, off: i64) -> i64 {
64 return (p[off] & 0xff) + ((p[off+1] & 0xff) << 8) + ((p[off+2] & 0xff) << 16) + ((p[off+3] & 0xff) << 24) }
65
66// pack ONE frame into MTU packets, written back-to-back into out as [pktlen:u16 LE][pkt]... .
67// returns the number of packets, or -1 if the frame exceeds 32 frags or out_cap (honest refusal).
68func dgm_pack_frame(kind: i64, sender: *u8, seq: i64, frame: *u8, n: i64, out: *u8, out_cap: i64) -> i64 {
69 if n <= 0 { return 0 - 1 }
70 let nf: i64 = (n + DGM_FRAG_CAP - 1) / DGM_FRAG_CAP
71 if nf > DGM_MAX_FRAGS { return 0 - 1 }
72 var o: i64 = 0
73 var f: i64 = 0
74 while f < nf {
75 var pl: i64 = n - f * DGM_FRAG_CAP
76 if pl > DGM_FRAG_CAP { pl = DGM_FRAG_CAP }
77 let pk: i64 = DGM_HDR + pl
78 if o + 2 + pk > out_cap { return 0 - 1 }
79 out[o] = (pk & 0xff) as u8
80 out[o+1] = ((pk >> 8) & 0xff) as u8
81 let b: i64 = o + 2
82 out[b] = 78 as u8 // 'N'
83 out[b+1] = 68 as u8 // 'D'
84 out[b+2] = (kind & 0xff) as u8
85 var i: i64 = 0
86 while i < 8 { out[b+3+i] = sender[i]; i = i + 1 }
87 dgm_wr_u32(out, b + 11, seq)
88 out[b+15] = (f & 0xff) as u8
89 out[b+16] = (nf & 0xff) as u8
90 out[b+17] = 0 as u8
91 var c: i64 = 0
92 while c < pl { out[b + DGM_HDR + c] = frame[f * DGM_FRAG_CAP + c]; c = c + 1 }
93 o = o + 2 + pk
94 f = f + 1
95 }
96 return nf }
97
98// parse+validate one packet header (rule 12: refuse anything malformed BEFORE touching state).
99// info: [0]=kind [1]=seq [2]=frag_i [3]=frag_n [4]=payload_off [5]=payload_len [6]=sender_off. -1 refused.
100func dgm_parse_pkt(pkt: *u8, n: i64, info: *i64) -> i64 {
101 if n < DGM_HDR + 1 { return 0 - 1 }
102 if pkt[0] != (78 as u8) { return 0 - 1 }
103 if pkt[1] != (68 as u8) { return 0 - 1 }
104 let fi: i64 = pkt[15] & 0xff
105 let fn: i64 = pkt[16] & 0xff
106 if fn < 1 { return 0 - 1 }
107 if fn > DGM_MAX_FRAGS { return 0 - 1 }
108 if fi >= fn { return 0 - 1 }
109 let pl: i64 = n - DGM_HDR
110 if pl > DGM_FRAG_CAP { return 0 - 1 }
111 info[0] = pkt[2] & 0xff
112 info[1] = dgm_rd_u32(pkt, 11)
113 info[2] = fi
114 info[3] = fn
115 info[4] = DGM_HDR
116 info[5] = pl
117 info[6] = 3
118 return 0 }
119
120func dgm_rx_init(mem: *u8) -> i64 {
121 var i: i64 = 0
122 while i < DGM_RX_REGION { mem[i] = 0 as u8; i = i + 1 }
123 return 0 }
124
125func dgm_sender_slot(mem: *u8, pkt: *u8, soff: i64) -> i64 {
126 var s: i64 = 0
127 while s < DGM_SENDERS {
128 let idp: *u8 = ((mem as i64) + s * 8) as *u8
129 var used: i64 = 0
130 var eq: i64 = 1
131 var i: i64 = 0
132 while i < 8 {
133 if idp[i] != (0 as u8) { used = 1 }
134 if idp[i] != pkt[soff + i] { eq = 0 }
135 i = i + 1
136 }
137 if used == 1 { if eq == 1 { return s } }
138 if used == 0 {
139 var j: i64 = 0
140 while j < 8 { idp[j] = pkt[soff + j]; j = j + 1 }
141 return s
142 }
143 s = s + 1
144 }
145 return 0 - 1 }
146
147// feed one received packet. returns: >0 = COMPLETE frame length written to out (rinfo[0]=kind,[1]=seq,
148// [2]=sender_slot), 0 = pending/duplicate/ignored, -1 = refused (malformed). Never reads past pkt[n].
149func dgm_rx_add(mem: *u8, pkt: *u8, n: i64, out: *u8, out_cap: i64, rinfo: *i64) -> i64 {
150 let info: *i64 = rinfo // reuse caller's box for the parse (12 i64 provided)
151 if dgm_parse_pkt(pkt, n, info) != 0 { return 0 - 1 }
152 let kind: i64 = info[0]
153 let seq: i64 = info[1]
154 let fi: i64 = info[2]
155 let fn: i64 = info[3]
156 let poff: i64 = info[4]
157 let plen: i64 = info[5]
158 let s: i64 = dgm_sender_slot(mem, pkt, info[6])
159 if s < 0 { return 0 }
160 let sb: i64 = 64 + s * DGM_SSTRIDE
161 let hip: *i64 = ((mem as i64) + sb) as *i64
162 let msk: *i64 = ((mem as i64) + sb + 8) as *i64
163 // dedupe: already delivered inside the 256-seq window?
164 if seq <= hip[0] { if hip[0] - seq < DGM_WIN {
165 if dgm_mask_test(msk, hip[0] - seq) == 1 { return 0 }
166 } else { return 0 } } // ancient (beyond the window) -> drop
167 // find/open the pending assembly for this seq
168 var pi: i64 = 0 - 1
169 var free_: i64 = 0 - 1
170 var old_i: i64 = 0
171 var old_seq: i64 = 0x7fffffffffffffff
172 var k: i64 = 0
173 while k < DGM_PENDING {
174 let pb: *i64 = ((mem as i64) + sb + 40 + k * DGM_PSTRIDE) as *i64
175 if pb[1] == 0 { if free_ < 0 { free_ = k } }
176 if pb[1] != 0 { if pb[0] == seq { pi = k } if pb[0] < old_seq { old_seq = pb[0]; old_i = k } }
177 k = k + 1
178 }
179 if pi < 0 {
180 if free_ >= 0 { pi = free_ } else {
181 if seq < old_seq { return 0 } // older than every pending -> not worth a slot
182 pi = old_i // stale eviction: newest wins the slot
183 }
184 let pb: *i64 = ((mem as i64) + sb + 40 + pi * DGM_PSTRIDE) as *i64
185 pb[0] = seq; pb[1] = fn; pb[2] = 0; pb[3] = 0; pb[4] = kind
186 }
187 let pb: *i64 = ((mem as i64) + sb + 40 + pi * DGM_PSTRIDE) as *i64
188 if pb[1] != fn { return 0 } // frag_n mismatch vs first-seen -> hostile/corrupt, ignore
189 if ((pb[2] >> fi) & 1) == 1 { return 0 } // duplicate fragment
190 pb[2] = pb[2] | (1 << fi)
191 pb[3] = pb[3] + 1
192 let lens: *i64 = ((mem as i64) + sb + 40 + pi * DGM_PSTRIDE + DGM_PFIX) as *i64
193 lens[fi] = plen
194 let buf: *u8 = ((mem as i64) + sb + 40 + pi * DGM_PSTRIDE + DGM_PFIX + DGM_PLENS) as *u8
195 var c: i64 = 0
196 while c < plen { buf[fi * DGM_FRAG_CAP + c] = pkt[poff + c]; c = c + 1 }
197 if pb[3] < fn { return 0 } // still assembling
198 // COMPLETE: stitch fragments in order
199 var total: i64 = 0
200 var f2: i64 = 0
201 while f2 < fn { total = total + lens[f2]; f2 = f2 + 1 }
202 if total > out_cap { pb[1] = 0; return 0 } // caller buffer too small -> drop assembly, honest 0
203 var o: i64 = 0
204 f2 = 0
205 while f2 < fn {
206 var c2: i64 = 0
207 while c2 < lens[f2] { out[o] = buf[f2 * DGM_FRAG_CAP + c2]; o = o + 1; c2 = c2 + 1 }
208 f2 = f2 + 1
209 }
210 // advance the delivered window (256-bit multi-word shift)
211 if seq > hip[0] {
212 dgm_mask_shift(msk, seq - hip[0])
213 msk[0] = msk[0] | 1
214 hip[0] = seq
215 } else {
216 dgm_mask_set(msk, hip[0] - seq)
217 }
218 pb[1] = 0 // free the slot
219 rinfo[0] = pb[4]
220 rinfo[1] = seq
221 rinfo[2] = s
222 return total }
223
224// list missing seqs for sender slot s inside the delivered window (below hi_seq) -> NACK candidates.
225func dgm_rx_gaps(mem: *u8, s: i64, out_seqs: *i64, max: i64) -> i64 {
226 let hip: *i64 = ((mem as i64) + 64 + s * DGM_SSTRIDE) as *i64
227 let msk: *i64 = ((mem as i64) + 64 + s * DGM_SSTRIDE + 8) as *i64
228 var cnt: i64 = 0
229 var b: i64 = 1
230 while b < DGM_WIN {
231 if cnt < max {
232 if dgm_mask_test(msk, b) == 0 {
233 let sq: i64 = hip[0] - b
234 if sq >= 0 { out_seqs[cnt] = sq; cnt = cnt + 1 }
235 }
236 }
237 b = b + 1
238 }
239 return cnt }
240
241// NACK packet: kind 0x4E, payload [cnt:1][seq:4]xcnt (single fragment; <= 236 seqs fits MTU easily).
242func dgm_pack_nack(sender: *u8, seqs: *i64, cnt: i64, out: *u8, out_cap: i64) -> i64 {
243 if cnt < 0 { return 0 - 1 }
244 if cnt > 32 { return 0 - 1 }
245 let pl: i64 = 1 + cnt * 4
246 let box: *u8 = out // build payload in place after the (single) header
247 if 2 + DGM_HDR + pl > out_cap { return 0 - 1 }
248 // reuse dgm_pack_frame with a stack payload: write payload into out tail first, then pack over it
249 let tmp: *u8 = ((out as i64) + out_cap - pl) as *u8
250 tmp[0] = (cnt & 0xff) as u8
251 var i: i64 = 0
252 while i < cnt { dgm_wr_u32(tmp, 1 + i * 4, seqs[i]); i = i + 1 }
253 return dgm_pack_frame(DGM_KIND_NACK, sender, 0, tmp, pl, box, out_cap - pl) }
254
255// parse a completed NACK frame body -> seq list. returns cnt or -1.
256func dgm_parse_nack(body: *u8, n: i64, out_seqs: *i64, max: i64) -> i64 {
257 if n < 1 { return 0 - 1 }
258 let cnt: i64 = body[0] & 0xff
259 if 1 + cnt * 4 > n { return 0 - 1 }
260 var i: i64 = 0
261 while i < cnt { if i < max { out_seqs[i] = dgm_rd_u32(body, 1 + i * 4) } i = i + 1 }
262 if cnt > max { return max }
263 return cnt }