nx_swarm_queue_gate.nx source

1// nx_swarm_queue_gate.nx -- gate for the intelligent queue scheduler. 2// T1 sq_should_admit table (PURE, the anti-oversubscribe + coordination core): 3// rank0/hold0/cap1/ok->admit · rank0/hold1/cap1/ok->wait(at-cap) · rank1/hold0/cap1/ok->wait 4// (someone ahead) · rank1/hold0/cap2/ok->admit · rank0/hold0/cap1/!ok->wait(telemetry). 5// T2 sq_rank fairness (PURE): higher-prio ahead->rank1 · lower-prio->rank0 · same-prio-earlier->rank1 6// · same-prio-later->rank0 · self excluded. 7// T3 sq_scan replay: crafted log with a LIVE holder (pid 1=init) + a DEAD-pid waiter (reaped) + 8// a HOLD tombstoned by DONE -> holders counts only the live untombstoned one; dead waiter absent. 9// T4 live: empty queue -> sq_wait_admit(self, huge cap) GRANTs immediately; release tombstones it. 10// license_tier: ORIGINAL expect_exit: 0 11import "nx_swarm_queue.nx" 12import "nx_gate_verdict.nx" 13 14func gq_write(path: *u8, s: *u8) -> i64 { 15 let fd: i64 = sys_openat_wr(path, 420) 16 if fd < 0 { return 0 - 1 } 17 sys_write(fd, s, std_slen(s)) 18 sys_close(fd) 19 return 0 20} 21 22func main(argc: i64, argv: *i64) -> i64 { 23 std_putln("SWARM-QUEUE-GATE: intelligent queuing + coordination (priority + FIFO + telemetry, no hard-deny)" as *u8) 24 var pass: i64 = 0 25 // T1 sq_should_admit table 26 let s1: i64 = sq_should_admit(0, 0, 1, 1) 27 let s2: i64 = sq_should_admit(0, 1, 1, 1) 28 let s3: i64 = sq_should_admit(1, 0, 1, 1) 29 let s4: i64 = sq_should_admit(1, 0, 2, 1) 30 let s5: i64 = sq_should_admit(0, 0, 1, 0) 31 if s1 == 1 && s2 == 0 && s3 == 0 && s4 == 1 && s5 == 0 { pass = pass + 1; std_putln("T1 PASS admit table (capacity + telemetry gate)" as *u8) } 32 if s1 != 1 || s2 != 0 || s3 != 0 || s4 != 1 || s5 != 0 { std_putln("T1 FAIL admit table" as *u8) } 33 // T2 sq_rank fairness -- craft waiter arrays: [other] 34 let wp: *i64 = sys_mmap(64) as *i64 35 let wpr: *i64 = sys_mmap(64) as *i64 36 let we: *i64 = sys_mmap(64) as *i64 37 // me = pid 100, prio 5, enq 1000 38 wp[0] = 200; wpr[0] = 9; we[0] = 500 // higher prio -> ahead 39 let r_hi: i64 = sq_rank(wp, wpr, we, 1, 100, 5, 1000) 40 wpr[0] = 2 // lower prio -> not ahead 41 let r_lo: i64 = sq_rank(wp, wpr, we, 1, 100, 5, 1000) 42 wpr[0] = 5; we[0] = 400 // same prio, earlier -> ahead 43 let r_eq_early: i64 = sq_rank(wp, wpr, we, 1, 100, 5, 1000) 44 we[0] = 1500 // same prio, later -> not ahead 45 let r_eq_late: i64 = sq_rank(wp, wpr, we, 1, 100, 5, 1000) 46 wp[0] = 100 // self excluded 47 let r_self: i64 = sq_rank(wp, wpr, we, 1, 100, 5, 1000) 48 if r_hi == 1 && r_lo == 0 && r_eq_early == 1 && r_eq_late == 0 && r_self == 0 { pass = pass + 1; std_putln("T2 PASS rank fairness (prio then FIFO, self excluded)" as *u8) } 49 if r_hi != 1 || r_lo != 0 || r_eq_early != 1 || r_eq_late != 0 || r_self != 0 { std_putln("T2 FAIL rank fairness" as *u8) } 50 // T3 sq_scan replay: pid 1 (init, alive) HOLD; dead pid WAIT; pid 1 later DONE? no -- test holder + dead-waiter reap. 51 let log: *u8 = "HOLD|1|5|4000|100\nWAIT|99999999|9|4000|50\nWAIT|1|3|1000|200\n" as *u8 52 // pid 1 last-state: HOLD then WAIT -> WAIT (waiter); dead 99999999 -> reaped. So holders=0, waiters=1 (pid1). 53 let sp: *i64 = sys_mmap(512) as *i64 54 let spr: *i64 = sys_mmap(512) as *i64 55 let se: *i64 = sys_mmap(512) as *i64 56 let snw: *i64 = sys_mmap(8) as *i64 57 let hold: i64 = sq_scan(log, std_slen(log), sp, spr, se, snw) 58 if hold == 0 && snw[0] == 1 && sp[0] == 1 { pass = pass + 1; std_putln("T3 PASS scan replay (latest-state per pid, dead reaped)" as *u8) } 59 if hold != 0 || snw[0] != 1 || sp[0] != 1 { std_puts("T3 FAIL hold="); std_pdec(hold); std_puts(" nw="); std_pdec(snw[0]); std_puts("\n" as *u8) } 60 // T4 live: fresh log, empty queue -> self admitted immediately. 61 // Reset = explicit vlen-0 put (`"" as *u8` is MISCOMPILED: points at the next literal -- see 62 // nx_empty_lit_probe). No lock unlink: unlinking a held flock file splits the mutex. 63 sov_put(SQ_STORE, "q:ids" as *u8, "x" as *u8, 0) 64 let me: i64 = sq_getpid() 65 // ⚠T4 WAS NOT HERMETIC AND FLAPPED WITH THE HOST'S MOOD (2026-07-30): it asserted an UNCONDITIONAL 66 // immediate GRANT, but sq_should_admit consults sq_telem_ok, which returns 0 whenever load1 >= ncores. 67 // Built on a quiet host it read 6/6; re-run at load 16.9 on 8 cores it read 5/6 -- and the SUBJECT was 68 // correct both times. A gate whose verdict depends on the machine's load is not measuring the subject, 69 // and a flapping gate is worse than no gate: it trains a reader to discount a real RED. 70 // ★So assert the CONTRACT the queue actually promises -- admission is CONDITIONAL -- by reading the 71 // telemetry first and requiring the matching branch. This is STRICTLY STRONGER than the old assertion: 72 // it now also proves the safety half, that a saturated host is REFUSED admission. 73 let t_ok: i64 = sq_telem_ok(10) 74 let g: i64 = sq_wait_admit(me, 5, 10, 1000000, 100, 50) 75 let rel: i64 = sq_release(me) 76 if t_ok == 1 { if g == 0 { pass = pass + 1; std_putln("T4 PASS telemetry OK -> empty-queue immediate GRANT + release" as *u8) } 77 if g != 0 { std_puts("T4 FAIL telemetry OK but not granted g="); std_pdec(g); std_puts("\n" as *u8) } } 78 if t_ok == 0 { if g != 0 { pass = pass + 1; std_putln("T4 PASS host saturated -> admission correctly REFUSED (load-aware safety half)" as *u8) } 79 if g == 0 { std_puts("T4 FAIL host saturated but ADMITTED anyway g="); std_pdec(g); std_puts("\n" as *u8) } } 80 sov_put(SQ_STORE, "q:ids" as *u8, "x" as *u8, 0) 81 // T5 BACKFILL decision table (PURE): front job in order -> GRANT-front(1); blocked (rank>0/at-cap) 82 // but SMALL + fits -> BACKFILL(2); blocked + BIG -> QUEUE(0); telemetry bad -> QUEUE regardless. 83 let bf_front: i64 = sq_admit_v2(0, 0, 1, 11000, 6000, 1) // rank0, cap1 -> front GRANT 84 let bf_small: i64 = sq_admit_v2(1, 0, 1, 4000, 6000, 1) // someone ahead, but I'm small+fit -> BACKFILL 85 let bf_big: i64 = sq_admit_v2(1, 0, 1, 11000, 6000, 1) // someone ahead + I'm BIG -> QUEUE 86 let bf_telem: i64 = sq_admit_v2(1, 0, 1, 4000, 6000, 0) // small but telemetry bad -> QUEUE 87 let bf_heavycap: i64 = sq_admit_v2(0, 1, 1, 11000, 6000, 1) // rank0 but a heavy holder fills cap -> big QUEUEs, not front 88 if bf_front == 1 && bf_small == 2 && bf_big == 0 && bf_telem == 0 && bf_heavycap == 0 { 89 pass = pass + 1; std_putln("T5 PASS backfill table (front-order / small-backfills / big-queues / telem-gates / heavy-cap)" as *u8) 90 } 91 if bf_front != 1 || bf_small != 2 || bf_big != 0 || bf_telem != 0 || bf_heavycap != 0 { std_putln("T5 FAIL backfill table" as *u8) } 92 // T6 heavy-holder count: a live BIG holder (pid 1, est 11000) counts as heavy; a small one (est 4000) does not. 93 let hlog: i64 = 0 94 let hb: *u8 = "HOLD|1|5|11000|100\nHOLD|1|5|11000|100\n" as *u8 95 let heavy1: i64 = sq_heavy_holders(hb, std_slen(hb), 6000) 96 let sb: *u8 = "HOLD|1|5|4000|100\n" as *u8 97 let heavy0: i64 = sq_heavy_holders(sb, std_slen(sb), 6000) 98 if heavy1 == 1 && heavy0 == 0 { pass = pass + 1; std_putln("T6 PASS heavy-holder classification (big counts, backfill-small does not)" as *u8) } 99 if heavy1 != 1 || heavy0 != 0 { std_puts("T6 FAIL heavy1="); std_pdec(heavy1); std_puts(" heavy0="); std_pdec(heavy0); std_puts("\n" as *u8) } 100 std_puts("SWARM-QUEUE-GATE pass=" as *u8) 101 std_pdec(pass) 102 // MIGRATED onto nx_gate_verdict by nx_gate_dry_apply (D001, minimal form): every check 103 // row above is untouched, so the PASS/FAIL vector cannot change; only the hand-rolled 104 // verdict emission is replaced by the ONE shared base class. Proven by nx_gate_migrate verify. 105 let ctr__dry: *i64 = gv_ctr() 106 ctr__dry[0] = pass 107 ctr__dry[1] = 6 108 let rc__dry: i64 = gv_verdict("SWARM-QUEUE-GATE" as *u8, ctr__dry, "teeth unchanged; verdict emission migrated onto the shared base class" as *u8) 109 sys_exit(rc__dry) 110 return rc__dry 111}