nx_dispatch_lease_gate.nx source

1// nx_dispatch_lease_gate.nx -- the REFEREE for WMS rung M4 (leasing dispatcher). 2// 3// module: nishi-core.autonomy.dispatch_lease_gate 4// capability: GATE (prints a VERDICT computed from REAL checks; incl negative control) 5// 6// PROVES (the gate requirement, exactly): 7// T1 picks-READY -- dl_pick_ready returns the highest-w READY rung, never blocked 8// T2 dispatch-leases -- dl_dispatch leases that rung (fd>=0, id matches, lease file exists) 9// T3 second-dispatch-DENIED -- a 2nd dl_try_lease on the SAME stream returns -1 (the 10// collision proof: two workers cannot hold one stream) 11// T3b keep-moving -- a fresh dl_dispatch while the top is leased returns the NEXT 12// ready stream (not the blocked one, not a stall) 13// T4 NEG CONTROL -- a deliberately BLOCKED stream (unmet dep) is NEVER dispatched: 14// dl_ready==0 AND no dispatch ever returns its id 15// T5 release-then-redispatch -- after dl_release the same stream leases again (fd>=0): 16// proves it is a LEASE (releasable), not a permanent mark 17// T6 TAMPER -- (a) fa_append oversize -> -2 (durability rejects, never torn) 18// (b) a stale lease (epoch older than DL_STALE_SEC) is STOLEN 19// (fd>=0) -- a dead worker's lease cannot wedge the machine 20// 21// VALIDITY (why the neg-control genuinely bites): if dl_pick_ready ignored an_deps_done it 22// would return G-BLOCKED -> T1/T4 RED. If dl_try_lease used plain open not O_EXCL, T3 would 23// return fd>=0 (collision) -> RED. If dl_release didn't unlink, T5 would stay -1 -> RED. 24// 25// HERMETIC: the gate WRITES its own scratch queue to /tmp/dlg_<ms>.tsv (9-field rows, same 26// grammar) and uses a fresh scratch lease dir /tmp/dlg_<ms>/ -- it NEVER touches production. 27// <ms> = sys_now_realtime_ms() so two concurrent gate runs never share lease state. 28// 29// WRITE DISCIPLINE: every evidence record (one buffer -> one locked fa_appendz) to 30// knowledge/status/dispatch_lease_gate.log -- eating our own dogfood (the torn-line floor). 31// license_tier: ORIGINAL 32import "nx_dispatch_lease.nx" 33import "nx_framed_append.nx" 34 35const GLOG: *u8 = "knowledge/status/dispatch_lease_gate.log\x00" as *u8 36 37func g_p(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} sys_write(1,s,n); return 0 } 38func g_n(v: i64) -> i64 { let bb: *u8=sys_mmap(28); var m: i64=v; if m<0{m=0-m; sys_write(1,"-" as *u8,1)}; let t: *u8=sys_mmap(28); var k: i64=0; if m==0{t[0]=48 as u8;k=1}; while m>0{t[k]=(48+(m%10)) as u8;m=m/10;k=k+1}; var i: i64=0; while i<k{bb[i]=t[k-1-i];i=i+1}; sys_write(1,bb,k); return 0 } 39 40// append decimal of v (NUL-terminated) into dst at off; returns new offset. 41func g_catd(dst: *u8, off: i64, v: i64) -> i64 { 42 let o: i64 = fa_catn(dst, off, v) 43 dst[o] = 0 as u8 44 return o 45} 46 47// one-buffer -> one locked fa_appendz evidence record: "MGATE test=<name> result=PASS|FAIL detail=<n>" 48func g_log(name: *u8, pass: i64, detail: i64) -> i64 { 49 let buf: *u8 = sys_mmap(256 + 16) 50 var o: i64 = 0 51 o = fa_cat(buf, o, "MGATE test=\x00" as *u8) 52 o = fa_cat(buf, o, name) 53 o = fa_cat(buf, o, " result=\x00" as *u8) 54 if pass == 1 { o = fa_cat(buf, o, "PASS\x00" as *u8) } else { o = fa_cat(buf, o, "FAIL\x00" as *u8) } 55 o = fa_cat(buf, o, " detail=\x00" as *u8) 56 o = fa_catn(buf, o, detail) 57 buf[o] = 0 as u8 58 return fa_appendz(GLOG, buf, 256) 59} 60 61// print + log one assertion; returns pass (1/0) so the caller can tally. 62func g_assert(name: *u8, pass: i64, detail: i64) -> i64 { 63 g_p(" " as *u8); g_p(name); g_p(": \x00" as *u8) 64 if pass == 1 { g_p("PASS\x00" as *u8) } else { g_p("FAIL\x00" as *u8) } 65 g_p(" (detail=\x00" as *u8); g_n(detail); g_p(")\n\x00" as *u8) 66 g_log(name, pass, detail) 67 return pass 68} 69 70// getpid (__syscall 39) is BROKEN on this backend (-25 for EVERY process -> not unique -> the real cause of the 71// intermittent-RED control-plane flake). Claim a process-unique i64 token ATOMICALLY via O_EXCL: first creator of 72// /tmp/nxuq_<n> owns <n>; a collision (EEXIST, fd<0) retries n+1. Seeded by monotonic us. Provably collision-free. 73func g_uniq() -> i64 { 74 var n: i64 = sys_now_us() 75 if n < 0 { n = 0 - n } 76 var tries: i64 = 0 77 while tries < 1000000 { 78 let p: *u8 = sys_mmap(64) 79 var o: i64 = fa_cat(p, 0, "/tmp/nxuq_\x00" as *u8) 80 o = fa_catn(p, o, n) 81 p[o] = 0 as u8 82 let fd: i64 = __syscall(SYS_OPENAT, AT_FDCWD, p, 0xc1, 0x1a4, 0, 0) // O_CREAT|O_EXCL|O_WRONLY 83 if fd >= 0 { sys_close(fd); return n } 84 n = n + 1 85 tries = tries + 1 86 } 87 return n 88} 89// build the scratch queue path /tmp/dlg_<ms>_<uniq>.tsv and the lease dir /tmp/dlg_<ms>_<uniq> into outs. 90func g_scratch_paths(ms: i64, qpath: *u8, ldir: *u8) -> i64 { 91 var o: i64 = 0 92 o = fa_cat(ldir, o, "/tmp/dlg_\x00" as *u8) 93 o = fa_catn(ldir, o, ms) 94 ldir[o] = 95 as u8; o = o + 1 // '_' -- ms-only collided at the same millisecond under concurrency (the flake); g_uniq makes scratch collision-proof 95 o = fa_catn(ldir, o, g_uniq()) 96 ldir[o] = 0 as u8 // ldir = "/tmp/dlg_<ms>_<uniq>" 97 var q: i64 = 0 98 while ldir[q] != (0 as u8) { qpath[q] = ldir[q]; q = q + 1 } 99 qpath[q] = 46 as u8; q = q + 1 // '.' 100 qpath[q] = 116 as u8; q = q + 1 // 't' 101 qpath[q] = 115 as u8; q = q + 1 // 's' 102 qpath[q] = 118 as u8; q = q + 1 // 'v' 103 qpath[q] = 0 as u8 // qpath = "/tmp/dlg_<ms>.tsv" 104 return 0 105} 106 107// write the controlled scratch fixture (9 tab fields per row; '\n' terminated). 108// Dep graph: G-READY-HI (w9, no dep), G-READY-LO (w5, no dep), G-BLOCKED (w9, dep 109// G-DEP which is still TODO not DONE -> blocked), G-DEP (w1, no dep, status TODO). 110// G-DEP is authored ONLY to be G-BLOCKED's unmet dependency. It MUST be TODO (so 111// G-BLOCKED stays blocked = the neg-control), but its weight is deliberately w1 -- 112// below G-READY-LO (w5) -- so that in the T3b keep-moving scan (G-READY-HI held) the 113// next-best READY UNLEASED row is unambiguously G-READY-LO, not G-DEP. (G-DEP is 114// itself independently ready since it has no deps; w1 keeps it from outranking the 115// intended keep-moving target without changing its TODO/blocked-dep role.) 116func g_write_fixture(path: *u8) -> i64 { 117 let fd: i64 = sys_openat_wr(path, 0x1a4) // O_CREAT|O_WRONLY|O_TRUNC, 0644 118 if fd < 0 { return 0 - 1 } 119 let s: *u8 = "G-READY-HI\tX\t9\tS\tBuilder\tTODO\t-\tgate\tready-hi\nG-READY-LO\tX\t5\tS\tBuilder\tTODO\t-\tgate\tready-lo\nG-BLOCKED\tX\t9\tS\tBuilder\tTODO\tG-DEP\tgate\tblocked-on-dep\nG-DEP\tX\t1\tS\tBuilder\tTODO\t-\tgate\tthe-unmet-dep\n\x00" as *u8 120 var n: i64 = 0 121 while s[n] != (0 as u8) { n = n + 1 } 122 sys_write(fd, s, n) 123 sys_close(fd) 124 return 0 125} 126 127// hand-write a STALE lease file (epoch=1, far older than DL_STALE_SEC) at <dir>/<id>.lease 128// so T6b can prove dl_try_lease STEALS it. 129func g_write_stale_lease(dir: *u8, id: *u8) -> i64 { 130 let lp: *u8 = sys_mmap(512) 131 dl_lease_path(dir, id, lp) 132 let fd: i64 = sys_openat_wr(lp, 0x1a4) 133 if fd < 0 { return 0 - 1 } 134 sys_write(fd, "1\x00" as *u8, 1) // epoch=1 (1970) -> always stale 135 sys_close(fd) 136 return 0 137} 138 139func main() -> i64 { 140 g_p("=== M4 DISPATCH-LEASE GATE: pick READY -> lease -> deny dup -> keep moving -> neg-ctrl blocked -> release-redispatch -> tamper ===\n\x00" as *u8) 141 let ms: i64 = sys_now_realtime_ms() 142 let qpath: *u8 = sys_mmap(128) 143 let ldir: *u8 = sys_mmap(128) 144 g_scratch_paths(ms, qpath, ldir) 145 sys_mkdir(ldir, 0x1ed) // fresh hermetic lease dir 146 if g_write_fixture(qpath) < 0 { g_p(" FIXTURE-WRITE-FAILED\n\x00" as *u8); sys_exit(1); return 1 } 147 148 let cx: *i64 = an_newcx() 149 let rows: i64 = an_load(qpath, cx) 150 if rows <= 0 { g_p(" FIXTURE-LOAD-FAILED\n\x00" as *u8); sys_exit(1); return 1 } 151 152 let rhi: i64 = an_find(cx, "G-READY-HI\x00" as *u8) 153 let rlo: i64 = an_find(cx, "G-READY-LO\x00" as *u8) 154 let rbl: i64 = an_find(cx, "G-BLOCKED\x00" as *u8) 155 var pass: i64 = 0 156 157 // ---- T1: picks the highest-w READY rung (G-READY-HI), skipping the blocked one ---- 158 let p1: i64 = dl_pick_ready(cx) 159 var t1: i64 = 0 160 if p1 == rhi { t1 = 1 } 161 pass = pass + g_assert("T1-picks-READY\x00" as *u8, t1, p1) 162 163 // ---- T2: dispatch leases that rung (fd>=0, id==G-READY-HI, lease file exists) ---- 164 let id_out: *u8 = sys_mmap(64) 165 let fd_out: *i64 = sys_mmap(16) as *i64 166 let fd2: i64 = dl_dispatch(cx, ldir, id_out, fd_out) 167 var t2: i64 = 0 168 if fd2 >= 0 { if dl_streq(id_out, "G-READY-HI\x00" as *u8) == 1 { if dl_lease_held(ldir, "G-READY-HI\x00" as *u8) == 1 { t2 = 1 } } } 169 pass = pass + g_assert("T2-dispatch-leases\x00" as *u8, t2, fd2) 170 171 // ---- T3: a SECOND lease attempt on the SAME held stream is DENIED (-1) ---- 172 let fd3: i64 = dl_try_lease(ldir, "G-READY-HI\x00" as *u8) 173 var t3: i64 = 0 174 if fd3 < 0 { t3 = 1 } // -1 = correctly denied (no collision) 175 if fd3 >= 0 { dl_release(ldir, "G-READY-HI\x00" as *u8, fd3) } // safety: don't leak a wrongly-won lease 176 pass = pass + g_assert("T3-second-dispatch-DENIED\x00" as *u8, t3, fd3) 177 178 // ---- T3b: keep-moving -- fresh dispatch while G-READY-HI is leased -> G-READY-LO ---- 179 let id_b: *u8 = sys_mmap(64) 180 let fd_b: *i64 = sys_mmap(16) as *i64 181 let fdb: i64 = dl_dispatch(cx, ldir, id_b, fd_b) 182 var t3b: i64 = 0 183 if fdb >= 0 { if dl_streq(id_b, "G-READY-LO\x00" as *u8) == 1 { t3b = 1 } } 184 pass = pass + g_assert("T3b-keep-moving\x00" as *u8, t3b, fdb) 185 186 // ---- T4: NEGATIVE CONTROL -- the BLOCKED stream is never ready, never dispatched ---- 187 // (a) dl_ready(G-BLOCKED)==0 because its dep G-DEP is TODO not DONE. 188 // (b) neither dispatched id (T2 id_out, T3b id_b) is ever G-BLOCKED. 189 var t4: i64 = 1 190 if dl_ready(cx, rbl) != 0 { t4 = 0 } // blocked must be NOT ready 191 if dl_streq(id_out, "G-BLOCKED\x00" as *u8) == 1 { t4 = 0 } 192 if dl_streq(id_b, "G-BLOCKED\x00" as *u8) == 1 { t4 = 0 } 193 pass = pass + g_assert("T4-NEGCTRL-blocked-never-dispatched\x00" as *u8, t4, dl_ready(cx, rbl)) 194 195 // ---- T5: release G-READY-HI, then it is re-dispatchable (lease again succeeds) ---- 196 dl_release(ldir, "G-READY-HI\x00" as *u8, fd_out[0]) 197 let fd5: i64 = dl_try_lease(ldir, "G-READY-HI\x00" as *u8) 198 var t5: i64 = 0 199 if fd5 >= 0 { t5 = 1 } // re-leasable after release (was -1 in T3) 200 if fd5 >= 0 { dl_release(ldir, "G-READY-HI\x00" as *u8, fd5) } // tidy up 201 pass = pass + g_assert("T5-release-then-redispatch\x00" as *u8, t5, fd5) 202 203 // ---- T6a: TAMPER -- fa_append oversize rejected (-2), durability never torn ---- 204 let ot: i64 = fa_append("/tmp/dlg_tamper.txt\x00" as *u8, "0123456789012345678901234567890123456789\x00" as *u8, 40, 8) 205 var t6a: i64 = 0 206 if ot == (0 - 2) { t6a = 1 } // -2 = oversized REJECTED 207 pass = pass + g_assert("T6a-TAMPER-oversize-rejected\x00" as *u8, t6a, ot) 208 209 // ---- T6b: TAMPER -- a STALE lease (epoch 1970) is STOLEN, not honored ---- 210 // release G-READY-LO first (T3b holds it), then plant a stale lease and steal it. 211 if fdb >= 0 { dl_release(ldir, "G-READY-LO\x00" as *u8, fd_b[0]) } 212 g_write_stale_lease(ldir, "G-STALE\x00" as *u8) 213 let fd6: i64 = dl_try_lease(ldir, "G-STALE\x00" as *u8) // must STEAL (epoch 1 < now-30) 214 var t6b: i64 = 0 215 if fd6 >= 0 { t6b = 1 } 216 if fd6 >= 0 { dl_release(ldir, "G-STALE\x00" as *u8, fd6) } 217 pass = pass + g_assert("T6b-TAMPER-stale-lease-stolen\x00" as *u8, t6b, fd6) 218 219 // ---- VERDICT ---- 220 let total: i64 = 8 // T1 T2 T3 T3b T4 T5 T6a T6b 221 var ok: i64 = 0 222 if pass == total { ok = 1 } 223 g_p("M4 DISPATCH-LEASE-GATE pass=\x00" as *u8); g_n(pass); g_p("/\x00" as *u8); g_n(total); g_p(" verdict=\x00" as *u8) 224 if ok == 1 { g_p("GREEN\n\x00" as *u8) } else { g_p("RED\n\x00" as *u8) } 225 226 // final verdict record -- one buffer -> one locked fa_appendz 227 let vb: *u8 = sys_mmap(256 + 16) 228 var vo: i64 = 0 229 vo = fa_cat(vb, vo, "DISPATCH-LEASE-GATE pass=\x00" as *u8) 230 vo = fa_catn(vb, vo, pass) 231 vo = fa_cat(vb, vo, "/\x00" as *u8) 232 vo = fa_catn(vb, vo, total) 233 vo = fa_cat(vb, vo, " epoch=\x00" as *u8) 234 vo = fa_catn(vb, vo, sys_now_realtime_sec()) 235 vo = fa_cat(vb, vo, " verdict=\x00" as *u8) 236 if ok == 1 { vo = fa_cat(vb, vo, "GREEN\x00" as *u8) } else { vo = fa_cat(vb, vo, "RED\x00" as *u8) } 237 vb[vo] = 0 as u8 238 fa_appendz(GLOG, vb, 256) 239 240 if ok == 1 { sys_exit(0); return 0 } 241 sys_exit(1) 242 return 1 243}