nx_swarm_beat.nx source
↩ module page · 462 lines · 18698 B
1// nx_swarm_beat.nx -- SWARM FABRIC fleet-beat store (SF-R1b): every worker reports its beacon NODE
2// row here; the hub reads back a fresh fleet snapshot. This is deliberately its OWN small organ per
3// [[feedback-no-risky-inplace-mgmt-api-deploy]] (NEVER bolt telemetry onto mgmt/hostctl) -- NAS-side
4// it runs via the pinned tools/call lane; nx_node_beacon (v2) emits the row, this organ stores it,
5// a later mgmt train may READ the snapshot (read-only swap of the hardcoded workers literal).
6//
7// nx_swarm_beat put <log> <snap> "<NODE row>" -- validate row; flock'd append to <log> (fa_append,
8// the R0b torn-write floor); atomically replace-by-name
9// in <snap> (whole-file rewrite, tmp+renameat)
10// nx_swarm_beat put "<NODE row>" -- NAS tools/call form: default paths swarm_beats.log +
11// node_rows.snap in CWD (tools daemon CWD = nishihost)
12// nx_swarm_beat show <snap> <window_sec> -- print FRESH/STALE per row (ts vs wall clock; the
13// window comes from the CALLER/conf -- no baked
14// freshness constant, rule 11)
15// nx_swarm_beat show <window_sec> -- default snap node_rows.snap
16// PATH LAW (least-authority BY CONSTRUCTION -- a leaked cap must not clobber nishihost binaries):
17// CLI paths must end .log/.snap, contain no "..", and be either bare basenames (land in CWD) or
18// live under /tmp/ (local tests). Everything else -> REFUSED before any io.
19// nx_swarm_beat [gate] -- self-gate: put/replace/show + neg-controls + the
20// fabricated-liveness liar-killer (an ancient or
21// future-forged ts can NEVER read FRESH)
22//
23// Row contract (v2 positional, additive -- emitted by nx_node_beacon):
24// NODE <name> <cpu> <load> <ncpu> <lpc> <memu> <mema> <memt> <conns> <ts_us> <temp_mc> <batt_pct>
25// v1 HONEST LIMIT: two DIFFERENT nodes putting at the same instant can lose one snap rewrite
26// (last-writer-wins whole-file); periodic beats self-heal next cycle and the LOG never loses a row
27// (flock'd). Serialize the rewrite under a lock file = v1.1 if beats ever collide in practice.
28// license_tier: ORIGINAL expect_exit:0
29import "nx_syscalls.nx"
30import "nx_framed_append.nx"
31const SB_MAGIC_60000000: i64 = 60000000
32const SB_MAGIC_1000000: i64 = 1000000
33const SB_MAGIC_7200000000: i64 = 7200000000
34const SB_MAGIC_3600: i64 = 3600
35const SB_MAGIC_600000000: i64 = 600000000
36
37const SB_SNAP_CAP: i64 = 65536
38const SB_ROW_CAP: i64 = 512
39
40func sb_puts(s: *u8) -> i64 { sys_write(1, s, fa_len(s)); return 0 }
41
42func sb_eq(a: *u8, b: *u8) -> i64 {
43 var i: i64 = 0
44 while a[i] != (0 as u8) { if a[i] != b[i] { return 0 } i = i + 1 }
45 if b[i] != (0 as u8) { return 0 }
46 return 1
47}
48
49// CLI path guard: must end ".log" or ".snap"; no ".."; bare basename (no '/') OR under /tmp/.
50func sb_path_ok(p: *u8) -> i64 {
51 let n: i64 = fa_len(p)
52 if n < 5 { return 0 }
53 // no ".." anywhere
54 var i: i64 = 0
55 while i + 1 < n {
56 if (p[i] as i64) == 46 { if (p[i + 1] as i64) == 46 { return 0 } }
57 i = i + 1
58 }
59 // extension .log or .snap
60 var ext_ok: i64 = 0
61 if n >= 4 {
62 if (p[n-4] as i64) == 46 { if (p[n-3] as i64) == 108 { if (p[n-2] as i64) == 111 { if (p[n-1] as i64) == 103 { ext_ok = 1 } } } }
63 }
64 if n >= 5 {
65 if (p[n-5] as i64) == 46 { if (p[n-4] as i64) == 115 { if (p[n-3] as i64) == 110 { if (p[n-2] as i64) == 97 { if (p[n-1] as i64) == 112 { ext_ok = 1 } } } } }
66 }
67 if ext_ok == 0 { return 0 }
68 // bare basename (no '/') is fine
69 var has_slash: i64 = 0
70 i = 0
71 while i < n { if (p[i] as i64) == 47 { has_slash = 1 } i = i + 1 }
72 if has_slash == 0 { return 1 }
73 // otherwise must start "/tmp/"
74 if n < 6 { return 0 }
75 if (p[0] as i64) != 47 { return 0 }
76 if (p[1] as i64) != 116 { return 0 }
77 if (p[2] as i64) != 109 { return 0 }
78 if (p[3] as i64) != 112 { return 0 }
79 if (p[4] as i64) != 47 { return 0 }
80 return 1
81}
82
83// bounded whole-file read; bytes read or -1 on open fail
84func sb_read(path: *u8, buf: *u8, cap: i64) -> i64 {
85 let fd: i64 = sys_openat_rd(path)
86 if fd < 0 { return 0 - 1 }
87 var tot: i64 = 0
88 while tot < cap {
89 let r: i64 = sys_read(fd, (buf as i64 + tot) as *u8, cap - tot)
90 if r <= 0 { break }
91 tot = tot + r
92 }
93 sys_close(fd)
94 return tot
95}
96
97// substring present? (n = haystack len)
98func sb_has(buf: *u8, n: i64, needle: *u8) -> i64 {
99 let m: i64 = fa_len(needle)
100 if m == 0 { return 0 }
101 var i: i64 = 0
102 while i + m <= n {
103 var k: i64 = 0
104 var hit: i64 = 1
105 while k < m { if buf[i + k] != needle[k] { hit = 0; k = m } else { k = k + 1 } }
106 if hit == 1 { return 1 }
107 i = i + 1
108 }
109 return 0
110}
111
112// signed decimal at p (within [p,n)); 1 ok / 0 no-digits; value -> vout[0], end -> pend[0]
113func sb_pint(buf: *u8, n: i64, p: i64, vout: *i64, pend: *i64) -> i64 {
114 var i: i64 = p
115 var neg: i64 = 0
116 if i < n { if (buf[i] as i64) == 45 { neg = 1; i = i + 1 } }
117 var v: i64 = 0
118 var d: i64 = 0
119 var go: i64 = 1
120 while go == 1 {
121 if i >= n { go = 0 } else {
122 let c: i64 = buf[i] as i64
123 if c >= 48 { if c <= 57 { v = v * 10 + (c - 48); d = d + 1; i = i + 1 } else { go = 0 } } else { go = 0 }
124 }
125 }
126 if d == 0 { return 0 }
127 if neg == 1 { v = 0 - v }
128 vout[0] = v
129 pend[0] = i
130 return 1
131}
132
133// validate one NODE row of exactly 13 tokens; name span -> nso/nlo, ts (token 10) -> tso. 1 ok / 0 bad.
134func sb_validate(row: *u8, n: i64, nso: *i64, nlo: *i64, tso: *i64) -> i64 {
135 if n < 6 { return 0 }
136 if row[0] != (78 as u8) { return 0 }
137 if row[1] != (79 as u8) { return 0 }
138 if row[2] != (68 as u8) { return 0 }
139 if row[3] != (69 as u8) { return 0 }
140 if row[4] != (32 as u8) { return 0 }
141 var i: i64 = 5
142 let ns: i64 = i
143 while i < n {
144 if (row[i] as i64) == 32 { break }
145 if (row[i] as i64) == 10 { return 0 }
146 i = i + 1
147 }
148 let nl: i64 = i - ns
149 if nl < 1 { return 0 }
150 if nl > 32 { return 0 }
151 let vout: *i64 = sys_mmap(16) as *i64
152 let pend: *i64 = sys_mmap(16) as *i64
153 var t: i64 = 2
154 var ts: i64 = 0 - 1
155 while t <= 12 {
156 if i >= n { return 0 }
157 if (row[i] as i64) != 32 { return 0 }
158 i = i + 1
159 let ok: i64 = sb_pint(row, n, i, vout, pend)
160 if ok == 0 { return 0 }
161 if t == 10 { ts = vout[0] }
162 i = pend[0]
163 t = t + 1
164 }
165 // after the last token: only end / newline allowed (strict 13-token contract)
166 if i < n {
167 if (row[i] as i64) != 10 { return 0 }
168 if i + 1 < n { return 0 }
169 }
170 nso[0] = ns
171 nlo[0] = nl
172 tso[0] = ts
173 return 1
174}
175
176// pure freshness verdict: 1 FRESH / 0 STALE. Future-forged ts (beyond 60s skew) is STALE too.
177func sb_fresh(ts: i64, now: i64, window_sec: i64) -> i64 {
178 if ts <= 0 { return 0 }
179 if ts > now + SB_MAGIC_60000000 { return 0 }
180 if now - ts < window_sec * SB_MAGIC_1000000 { return 1 }
181 return 0
182}
183
184// replace-by-name in snap: keep every line whose name differs, append the new row, tmp+renameat.
185func sb_snap_replace(snap: *u8, row: *u8, rowlen: i64, ns: i64, nl: i64) -> i64 {
186 let old: *u8 = sys_mmap(SB_SNAP_CAP)
187 var on: i64 = sb_read(snap, old, SB_SNAP_CAP)
188 if on < 0 { on = 0 }
189 let neu: *u8 = sys_mmap(SB_SNAP_CAP + SB_ROW_CAP)
190 var o: i64 = 0
191 var i: i64 = 0
192 while i < on {
193 var e: i64 = i
194 while e < on { if (old[e] as i64) == 10 { break } e = e + 1 }
195 var keep: i64 = 1
196 if e - i >= 6 {
197 if old[i] == (78 as u8) {
198 let ls: i64 = i + 5
199 var le: i64 = ls
200 while le < e { if (old[le] as i64) == 32 { break } le = le + 1 }
201 if le - ls == nl {
202 var k: i64 = 0
203 var same: i64 = 1
204 while k < nl { if old[ls + k] != row[ns + k] { same = 0; k = nl } else { k = k + 1 } }
205 if same == 1 { keep = 0 }
206 }
207 }
208 }
209 if keep == 1 {
210 if e > i {
211 var c: i64 = i
212 while c < e { neu[o] = old[c]; o = o + 1; c = c + 1 }
213 neu[o] = 10 as u8
214 o = o + 1
215 }
216 }
217 i = e + 1
218 }
219 var k2: i64 = 0
220 while k2 < rowlen { neu[o] = row[k2]; o = o + 1; k2 = k2 + 1 }
221 neu[o] = 10 as u8
222 o = o + 1
223 let tmp: *u8 = sys_mmap(512)
224 var to: i64 = 0
225 to = fa_cat(tmp, to, snap)
226 to = fa_cat(tmp, to, "." as *u8)
227 to = fa_catn(tmp, to, sys_now_us())
228 to = fa_cat(tmp, to, ".tmp" as *u8)
229 tmp[to] = 0 as u8
230 let fd: i64 = sys_openat_wr(tmp, 0x1a4)
231 if fd < 0 { return 0 - 4 }
232 let wr: i64 = fa_write_all(fd, neu, o)
233 sys_close(fd)
234 if wr != o { return 0 - 4 }
235 let rr: i64 = sys_renameat(tmp, snap)
236 if rr < 0 { return 0 - 4 }
237 return 0
238}
239
240// the whole put: validate -> flock'd log append -> snap replace. 0 ok / -3 validate / -4 io.
241func sb_put(log: *u8, snap: *u8, row: *u8) -> i64 {
242 let n: i64 = fa_len(row)
243 if n >= SB_ROW_CAP { return 0 - 3 }
244 let nso: *i64 = sys_mmap(16) as *i64
245 let nlo: *i64 = sys_mmap(16) as *i64
246 let tso: *i64 = sys_mmap(16) as *i64
247 let ok: i64 = sb_validate(row, n, nso, nlo, tso)
248 if ok == 0 { return 0 - 3 }
249 let rec: *u8 = sys_mmap(SB_ROW_CAP + 8)
250 var o: i64 = 0
251 var i: i64 = 0
252 while i < n { if (row[i] as i64) != 10 { rec[o] = row[i]; o = o + 1 } i = i + 1 }
253 let ar: i64 = fa_append(log, rec, o, SB_ROW_CAP + 8) // fa_append adds the '\n' frame; returns o+1
254 if ar != o + 1 { return 0 - 4 }
255 return sb_snap_replace(snap, rec, o, nso[0], nlo[0])
256}
257
258// show: verdict per row + tallies. 0 always (report tool).
259func sb_show(snap: *u8, window_sec: i64) -> i64 {
260 let buf: *u8 = sys_mmap(SB_SNAP_CAP)
261 let n: i64 = sb_read(snap, buf, SB_SNAP_CAP)
262 let now: i64 = sys_now_us()
263 var fresh: i64 = 0
264 var stale: i64 = 0
265 var bad: i64 = 0
266 var i: i64 = 0
267 while i < n {
268 var e: i64 = i
269 while e < n { if (buf[e] as i64) == 10 { break } e = e + 1 }
270 if e > i {
271 let nso: *i64 = sys_mmap(16) as *i64
272 let nlo: *i64 = sys_mmap(16) as *i64
273 let tso: *i64 = sys_mmap(16) as *i64
274 let ok: i64 = sb_validate((buf as i64 + i) as *u8, e - i, nso, nlo, tso)
275 if ok == 1 {
276 let fr: i64 = sb_fresh(tso[0], now, window_sec)
277 if fr == 1 { fresh = fresh + 1; sb_puts("FRESH " as *u8) } else { stale = stale + 1; sb_puts("STALE " as *u8) }
278 } else {
279 bad = bad + 1
280 sb_puts("BAD " as *u8)
281 }
282 sys_write(1, (buf as i64 + i) as *u8, e - i)
283 sb_puts("\n" as *u8)
284 }
285 i = e + 1
286 }
287 sb_puts("SWARMBEAT-SHOW fresh=" as *u8)
288 let t: *u8 = sys_mmap(64)
289 var to: i64 = 0
290 to = fa_catn(t, to, fresh)
291 to = fa_cat(t, to, " stale=" as *u8)
292 to = fa_catn(t, to, stale)
293 to = fa_cat(t, to, " bad=" as *u8)
294 to = fa_catn(t, to, bad)
295 to = fa_cat(t, to, "\n" as *u8)
296 sys_write(1, t, to)
297 return 0
298}
299
300// build a synthetic v2 row for the gate
301func sb_mkrow(dst: *u8, name: *u8, cpu: i64, ts: i64) -> i64 {
302 var o: i64 = 0
303 o = fa_cat(dst, o, "NODE " as *u8)
304 o = fa_cat(dst, o, name)
305 o = fa_cat(dst, o, " " as *u8); o = fa_catn(dst, o, cpu)
306 o = fa_cat(dst, o, " 500 4 125 40 8000 16000 0 " as *u8)
307 o = fa_catn(dst, o, ts)
308 o = fa_cat(dst, o, " -1 92" as *u8)
309 dst[o] = 0 as u8
310 return o
311}
312
313func sb_gate() -> i64 {
314 var pass: i64 = 0
315 var total: i64 = 0
316 let now: i64 = sys_now_us()
317 let log: *u8 = sys_mmap(256)
318 var lo: i64 = 0
319 lo = fa_cat(log, lo, "/tmp/swb_gate_" as *u8)
320 lo = fa_catn(log, lo, now)
321 lo = fa_cat(log, lo, ".log" as *u8)
322 log[lo] = 0 as u8
323 let snap: *u8 = sys_mmap(256)
324 var so: i64 = 0
325 so = fa_cat(snap, so, "/tmp/swb_gate_" as *u8)
326 so = fa_catn(snap, so, now)
327 so = fa_cat(snap, so, ".snap" as *u8)
328 snap[so] = 0 as u8
329 let row: *u8 = sys_mmap(SB_ROW_CAP)
330 let buf: *u8 = sys_mmap(SB_SNAP_CAP)
331
332 // T1 first put lands
333 total = total + 1
334 sb_mkrow(row, "alpha" as *u8, 11, now)
335 var r: i64 = sb_put(log, snap, row)
336 var n: i64 = sb_read(snap, buf, SB_SNAP_CAP)
337 if r == 0 { if sb_has(buf, n, "NODE alpha 11 " as *u8) == 1 { pass = pass + 1; sb_puts("T1 put-lands OK\n" as *u8) } }
338
339 // T2 second node adds a row (both present)
340 total = total + 1
341 sb_mkrow(row, "bravo" as *u8, 22, now)
342 r = sb_put(log, snap, row)
343 n = sb_read(snap, buf, SB_SNAP_CAP)
344 if r == 0 { if sb_has(buf, n, "NODE alpha 11 " as *u8) == 1 { if sb_has(buf, n, "NODE bravo 22 " as *u8) == 1 { pass = pass + 1; sb_puts("T2 two-nodes OK\n" as *u8) } } }
345
346 // T3 same-name put REPLACES (old bytes gone, new present, bravo intact)
347 total = total + 1
348 sb_mkrow(row, "alpha" as *u8, 77, now)
349 r = sb_put(log, snap, row)
350 n = sb_read(snap, buf, SB_SNAP_CAP)
351 if r == 0 { if sb_has(buf, n, "NODE alpha 77 " as *u8) == 1 { if sb_has(buf, n, "NODE alpha 11 " as *u8) == 0 { if sb_has(buf, n, "NODE bravo 22 " as *u8) == 1 { pass = pass + 1; sb_puts("T3 replace OK\n" as *u8) } } } }
352
353 // T4 NEG: 12-token row refused, snap unchanged
354 total = total + 1
355 let before: i64 = sb_read(snap, buf, SB_SNAP_CAP)
356 var o4: i64 = 0
357 o4 = fa_cat(row, o4, "NODE mallory 1 2 3 4 5 6 7 8 " as *u8)
358 o4 = fa_catn(row, o4, now)
359 row[o4] = 0 as u8
360 r = sb_put(log, snap, row)
361 n = sb_read(snap, buf, SB_SNAP_CAP)
362 if r == (0 - 3) { if n == before { if sb_has(buf, n, "mallory" as *u8) == 0 { pass = pass + 1; sb_puts("T4 neg-short-row REFUSED\n" as *u8) } } }
363
364 // T5 NEG: wrong prefix refused
365 total = total + 1
366 sb_mkrow(row, "carol" as *u8, 33, now)
367 row[0] = 88 as u8
368 r = sb_put(log, snap, row)
369 n = sb_read(snap, buf, SB_SNAP_CAP)
370 if r == (0 - 3) { if sb_has(buf, n, "carol" as *u8) == 0 { pass = pass + 1; sb_puts("T5 neg-prefix REFUSED\n" as *u8) } }
371
372 // T6 LIAR-KILLER: ancient ts can never read FRESH; fresh row stays FRESH
373 total = total + 1
374 let old_ts: i64 = now - SB_MAGIC_7200000000
375 sb_mkrow(row, "ghost" as *u8, 44, old_ts)
376 r = sb_put(log, snap, row)
377 var lk: i64 = 0
378 if r == 0 { if sb_fresh(old_ts, now, SB_MAGIC_3600) == 0 { if sb_fresh(now, now, SB_MAGIC_3600) == 1 { lk = 1 } } }
379 if lk == 1 { pass = pass + 1; sb_puts("T6 liar-killer stale-never-fresh OK\n" as *u8) }
380
381 // T7 LIAR-KILLER: future-forged ts is STALE
382 total = total + 1
383 if sb_fresh(now + SB_MAGIC_600000000, now, SB_MAGIC_3600) == 0 { pass = pass + 1; sb_puts("T7 liar-killer future-forged-stale OK\n" as *u8) }
384
385 // T8 show smoke over the real snap (3 rows: alpha/bravo fresh, ghost stale)
386 total = total + 1
387 sb_puts("--- show (window 3600s) ---\n" as *u8)
388 sb_show(snap, SB_MAGIC_3600)
389 pass = pass + 1
390
391 // T9 PATH LAW: traversal + non-tmp absolute + wrong extension all refused
392 total = total + 1
393 var p9: i64 = 1
394 if sb_path_ok("../x.snap" as *u8) != 0 { p9 = 0 }
395 if sb_path_ok("/etc/x.snap" as *u8) != 0 { p9 = 0 }
396 if sb_path_ok("/tmp/evil.elf" as *u8) != 0 { p9 = 0 }
397 if sb_path_ok("sites.elf" as *u8) != 0 { p9 = 0 }
398 if p9 == 1 { pass = pass + 1; sb_puts("T9 path-law refusals OK\n" as *u8) }
399
400 // T10 PATH LAW: legit forms accepted (bare basename + /tmp)
401 total = total + 1
402 var p10: i64 = 1
403 if sb_path_ok("node_rows.snap" as *u8) != 1 { p10 = 0 }
404 if sb_path_ok("swarm_beats.log" as *u8) != 1 { p10 = 0 }
405 if sb_path_ok(snap) != 1 { p10 = 0 }
406 if p10 == 1 { pass = pass + 1; sb_puts("T10 path-law accepts OK\n" as *u8) }
407
408 let t: *u8 = sys_mmap(128)
409 var to: i64 = 0
410 to = fa_cat(t, to, "SWARMBEATGATE " as *u8)
411 to = fa_catn(t, to, pass)
412 to = fa_cat(t, to, "/" as *u8)
413 to = fa_catn(t, to, total)
414 if pass == total { to = fa_cat(t, to, " verdict=GREEN\n" as *u8) } else { to = fa_cat(t, to, " verdict=RED\n" as *u8) }
415 sys_write(1, t, to)
416 if pass == total { return 0 }
417 return 1
418}
419
420func main(argc: i64, argv: *i64) -> i64 {
421 if argc >= 2 {
422 let verb: *u8 = argv[1] as *u8
423 if sb_eq(verb, "put" as *u8) == 1 {
424 var log: *u8 = "swarm_beats.log" as *u8
425 var snap: *u8 = "node_rows.snap" as *u8
426 var row: *u8 = 0 as *u8
427 if argc == 3 { row = argv[2] as *u8 }
428 if argc >= 5 { log = argv[2] as *u8; snap = argv[3] as *u8; row = argv[4] as *u8 }
429 if (row as i64) == 0 { sb_puts("usage: nx_swarm_beat put [<log> <snap>] <NODE-row>\n" as *u8); return 2 }
430 if sb_path_ok(log) == 0 { sb_puts("SWARMBEAT put REFUSED (path law)\n" as *u8); return 3 }
431 if sb_path_ok(snap) == 0 { sb_puts("SWARMBEAT put REFUSED (path law)\n" as *u8); return 3 }
432 let r: i64 = sb_put(log, snap, row)
433 if r == 0 { sb_puts("SWARMBEAT put OK\n" as *u8); return 0 }
434 if r == (0 - 3) { sb_puts("SWARMBEAT put REFUSED (row contract)\n" as *u8); return 3 }
435 sb_puts("SWARMBEAT put IO-FAIL\n" as *u8)
436 return 4
437 }
438 if sb_eq(verb, "now" as *u8) == 1 {
439 // expose the fabric clock (sys_now_us) so tooling/tests can stamp rows in the SAME
440 // units the freshness check uses -- date-epoch does NOT match this clock.
441 let t: *u8 = sys_mmap(32)
442 var o: i64 = fa_catn(t, 0, sys_now_us())
443 t[o] = 10 as u8
444 sys_write(1, t, o + 1)
445 return 0
446 }
447 if sb_eq(verb, "show" as *u8) == 1 {
448 var snap2: *u8 = "node_rows.snap" as *u8
449 var ws: *u8 = 0 as *u8
450 if argc == 3 { ws = argv[2] as *u8 }
451 if argc >= 4 { snap2 = argv[2] as *u8; ws = argv[3] as *u8 }
452 if (ws as i64) == 0 { sb_puts("usage: nx_swarm_beat show [<snap>] <window_sec>\n" as *u8); return 2 }
453 if sb_path_ok(snap2) == 0 { sb_puts("SWARMBEAT show REFUSED (path law)\n" as *u8); return 3 }
454 let vout: *i64 = sys_mmap(16) as *i64
455 let pend: *i64 = sys_mmap(16) as *i64
456 let ok: i64 = sb_pint(ws, fa_len(ws), 0, vout, pend)
457 if ok == 0 { return 2 }
458 return sb_show(snap2, vout[0])
459 }
460 }
461 return sb_gate()
462}