nx_swarm_tend_gate.nx source
↩ module page · 92 lines · 6286 B
1// nx_swarm_tend_gate.nx -- gate for the active queue tender (SOTA queue ops, no fire-and-forget).
2// T1 PURE verdict table: no waiters->OK, fresh waiter->DRAINING, starved waiter->JAMMED.
3// T2 ACTIVE REAP: plant a HOLD for a DEAD pid + a HOLD for a LIVE pid (init=1) in the store, with
4// NO waiter present; tend_reap frees ONLY the dead one (writes DONE) -> the leaked-slot bug the
5// passive path can't fix (no waiter to trigger it). reaped==1; live holder untouched.
6// T3 JAM DETECTION live: plant a waiter with an ancient enqueue timestamp -> tend_health sees a
7// large oldest_wait_ms -> verdict JAMMED (the traffic-jam alarm fires before a silent timeout).
8// T4 clean: fresh empty queue -> OK, 0 reaped.
9// T5 MULTI-ENTRY index: dead holder + live holder coexist in q:ids; reap frees ONLY the dead one,
10// the live one stays counted. (The case a false "seg_store bug" diagnosis claimed impossible --
11// the real bugs were nx_cc empty-literal + const-syscall getpid, both fixed/worked around.)
12// license_tier: ORIGINAL expect_exit: 0
13import "nx_swarm_tend.nx"
14import "nx_gate_verdict.nx"
15
16func tg_put(pid: i64, ty: *u8, prio: i64, est: i64, us: i64) -> i64 {
17 let row: *u8 = sys_mmap(256) as *u8
18 sq_mkrow(ty, pid, prio, est, us, row)
19 sq_put_state(pid, row)
20 sq_add_id(pid)
21 return 0
22}
23
24func main(argc: i64, argv: *i64) -> i64 {
25 std_putln("SWARM-TEND-GATE: active queue tending (reap leaked holders + detect jams, not fire-and-forget)" as *u8)
26 var pass: i64 = 0
27 // T1 verdict table
28 let vo: i64 = tend_verdict(0, 0, 300000)
29 let vd: i64 = tend_verdict(2, 1000, 300000)
30 let vj: i64 = tend_verdict(1, 999999, 300000)
31 if vo == TEND_OK && vd == TEND_DRAINING && vj == TEND_JAMMED { pass = pass + 1; std_putln("T1 PASS verdict table (OK/DRAINING/JAMMED)" as *u8) }
32 if vo != TEND_OK || vd != TEND_DRAINING || vj != TEND_JAMMED { std_putln("T1 FAIL verdict table" as *u8) }
33 // the gate's OWN pid = guaranteed-alive (own /proc/<pid>/comm readable; pid 1 is restricted in WSL)
34 let me: i64 = sq_getpid()
35 // RESET IDIOM: explicit vlen=0 put ("x" is a donor pointer, NEVER dereferenced at len 0). The
36 // obvious sov_put_str(store, key, "") is a LANDMINE: `"" as *u8` is MISCOMPILED (emits no NUL;
37 // the pointer lands ON the next rodata literal -- witnessed by nx_empty_lit_probe: strlen("")==4
38 // aliasing "HOLD"). That single bug caused every prior RED here (q:ids reset wrote "HOLD").
39 // NO lock unlink: unlinking a lock file another process holds an flock on SPLITS the mutex
40 // (old holder keeps the dead inode, new lockers lock a fresh one -> both "hold" it).
41 // T2 active reap of a leaked (dead) holder -- the core value: a dead HOLD gets DONE'd WITHOUT a
42 // waiter polling (the leaked-slot bug the passive path can't fix).
43 sov_put(SQ_STORE, "q:ids" as *u8, "x" as *u8, 0)
44 tg_put(88888888, "HOLD" as *u8, 5, 4000, sys_now_us()) // DEAD holder (leaked slot), sole entry
45 let reaped: i64 = tend_reap()
46 let hbuf: *i64 = sys_mmap(8) as *i64
47 let obuf: *i64 = sys_mmap(8) as *i64
48 let waiters: i64 = tend_health(hbuf, obuf)
49 // after reap: dead holder DONE (not counted), 0 live holders, 0 waiters
50 if reaped == 1 && hbuf[0] == 0 && waiters == 0 { pass = pass + 1; std_putln("T2 PASS active reap: leaked dead holder freed WITHOUT a waiter (passive path can't)" as *u8) }
51 if reaped != 1 || hbuf[0] != 0 || waiters != 0 { std_puts("T2 FAIL reaped="); std_pdec(reaped); std_puts(" holders="); std_pdec(hbuf[0]); std_puts(" waiters="); std_pdec(waiters); std_puts("\n" as *u8) }
52 // T3 jam detection: a LIVE ancient waiter (sole entry) -> JAMMED
53 sov_put(SQ_STORE, "q:ids" as *u8, "x" as *u8, 0)
54 let ancient: i64 = sys_now_us() - 600000000 // ~600s ago
55 tg_put(me, "WAIT" as *u8, 5, 4000, ancient) // live waiter (this process), enqueued long ago
56 let v3: i64 = tend_run()
57 if v3 == TEND_JAMMED { pass = pass + 1; std_putln("T3 PASS jam detection (live ancient waiter -> JAMMED alarm before a silent timeout)" as *u8) }
58 if v3 != TEND_JAMMED { std_puts("T3 FAIL v3="); std_pdec(v3); std_puts("\n" as *u8) }
59 // T4 clean empty queue (this final reset also REPAIRS the production q:ids the buggy resets poisoned)
60 sov_put(SQ_STORE, "q:ids" as *u8, "x" as *u8, 0)
61 let hb2: *i64 = sys_mmap(8) as *i64
62 let ob2: *i64 = sys_mmap(8) as *i64
63 let w4: i64 = tend_health(hb2, ob2)
64 let v4: i64 = tend_verdict(w4, ob2[0], TEND_JAM_MS)
65 let rp4: i64 = tend_reap()
66 if v4 == TEND_OK && w4 == 0 && rp4 == 0 { pass = pass + 1; std_putln("T4 PASS clean empty queue -> OK, 0 reaped" as *u8) }
67 if v4 != TEND_OK || w4 != 0 || rp4 != 0 { std_putln("T4 FAIL clean" as *u8) }
68 // T5 multi-entry index: dead + live holders TOGETHER; reap kills only the dead, live stays counted.
69 sov_put(SQ_STORE, "q:ids" as *u8, "x" as *u8, 0)
70 tg_put(88888888, "HOLD" as *u8, 5, 4000, sys_now_us())
71 tg_put(me, "HOLD" as *u8, 5, 4000, sys_now_us())
72 let rp5: i64 = tend_reap()
73 let hb5: *i64 = sys_mmap(8) as *i64
74 let ob5: *i64 = sys_mmap(8) as *i64
75 let w5: i64 = tend_health(hb5, ob5)
76 if rp5 == 1 && hb5[0] == 1 && w5 == 0 { pass = pass + 1; std_putln("T5 PASS multi-entry index: dead reaped, live holder kept+counted (2 pids in q:ids)" as *u8) }
77 if rp5 != 1 || hb5[0] != 1 || w5 != 0 { std_puts("T5 FAIL rp="); std_pdec(rp5); std_puts(" holders="); std_pdec(hb5[0]); std_puts(" w="); std_pdec(w5); std_puts("\n" as *u8) }
78 // leave production clean: my HOLD -> DONE, index reset empty
79 sq_release(me)
80 sov_put(SQ_STORE, "q:ids" as *u8, "x" as *u8, 0)
81 std_puts("SWARM-TEND-GATE pass=" as *u8)
82 std_pdec(pass)
83 // MIGRATED onto nx_gate_verdict by nx_gate_dry_apply (D001, minimal form): every check
84 // row above is untouched, so the PASS/FAIL vector cannot change; only the hand-rolled
85 // verdict emission is replaced by the ONE shared base class. Proven by nx_gate_migrate verify.
86 let ctr__dry: *i64 = gv_ctr()
87 ctr__dry[0] = pass
88 ctr__dry[1] = 5
89 let rc__dry: i64 = gv_verdict("SWARM-TEND-GATE" as *u8, ctr__dry, "teeth unchanged; verdict emission migrated onto the shared base class" as *u8)
90 sys_exit(rc__dry)
91 return rc__dry
92}