nx_ws_cas_gate.nx source

1// nx_ws_cas_gate.nx -- the REFEREE for WMS-R8 (concurrent-SSOT-safe registry write). 2// 3// Proves ws_put_locked defeats the concurrent-writer LOST-UPDATE race, AND proves the test can 4// DETECT lost updates (the mandatory negative control): 5// GOOD lane: NWORKERS fork, each ws_put_locked's NPER distinct keys into a FRESH /tmp store. 6// After join, ALL NWORKERS*NPER keys MUST be retrievable -> ZERO lost updates. 7// BAD lane (NEG-CONTROL): the SAME concurrency, SAME distinct keys, but the UNLOCKED ws_put_p -> 8// concurrent segid + manifest.txt RMW collisions LOSE updates -> retrievable < total. 9// If the bad lane loses NOTHING, the detector is worthless -> RED (no false green). 10// Each lane uses its OWN per-run unique prefix under /tmp (epoch-stamped) so every run starts on a 11// pristine store -- /tmp is disposable, so no clearing/unlink is needed (mirrors the R0 gate's 12// /tmp/fa_*.log discipline). Verification uses ss_open ONCE + ss_hget per key (zero per-call IO, 13// the same snapshot discipline the R4 audit was hardened to). The evidence line is one assembled 14// record written via fa_appendz -> it DOGFOODS the restored R0b lock-framed primitive (no torn 15// self-log -- closing the gap the critic flagged on earlier gates). Evidence -> 16// knowledge/status/ws_cas_gate.log. GREEN (exit 0) iff good==total AND bad<total; else RED (exit 1). 17// Sovereign: nx_ws_cas + nx_workstream_store + nx_seg_store + nx_framed_append + nx_syscalls. 18// license_tier: ORIGINAL 19import "nx_ws_cas.nx" 20import "nx_workstream_store.nx" 21import "nx_seg_store.nx" 22import "nx_framed_append.nx" 23import "nx_syscalls.nx" 24 25const CAS_NWORKERS: i64 = 12 26const CAS_NPER: i64 = 12 // 12*12 = 144 commits/lane (< the 256 manifest cap) 27const CAS_REC_CAP: i64 = 256 28const CAS_LOG: *u8 = "knowledge/status/ws_cas_gate.log" 29 30// stdout-only writers (the human report; the durable channel is the fa_appendz log line) 31func so(s: *u8) -> i64 { 32 var n: i64 = 0 33 while s[n] != (0 as u8) { n = n + 1 } 34 sys_write(1, s, n) 35 return 0 36} 37func son(v: i64) -> i64 { 38 let bb: *u8 = sys_mmap(28) 39 var m: i64 = v 40 if m < 0 { sys_write(1, "-" as *u8, 1); m = 0 - m } 41 let t: *u8 = sys_mmap(28) 42 var k: i64 = 0 43 if m == 0 { t[0] = 48 as u8; k = 1 } 44 while m > 0 { t[k] = (48 + (m % 10)) as u8; m = m / 10; k = k + 1 } 45 var i: i64 = 0 46 while i < k { bb[i] = t[k - 1 - i]; i = i + 1 } 47 sys_write(1, bb, k) 48 return 0 49} 50 51// append NUL-term s into dst at off -> new off 52func cg_cat(dst: *u8, off: i64, s: *u8) -> i64 { 53 var i: i64 = 0 54 while s[i] != (0 as u8) { dst[off + i] = s[i]; i = i + 1 } 55 return off + i 56} 57// append decimal of v into dst at off -> new off 58func cg_catn(dst: *u8, off: i64, v: i64) -> i64 { 59 var m: i64 = v 60 var o: i64 = off 61 if m == 0 { dst[o] = 48 as u8; return o + 1 } 62 let t: *u8 = sys_mmap(28) 63 var k: i64 = 0 64 while m > 0 { t[k] = (48 + (m % 10)) as u8; m = m / 10; k = k + 1 } 65 var i: i64 = 0 66 while i < k { dst[o + i] = t[k - 1 - i]; i = i + 1 } 67 return o + k 68} 69 70// build the deterministic test key for (w,i) into out (NUL-term): "k:w<w>i<i>" 71func cg_key(out: *u8, w: i64, i: i64) -> i64 { 72 var o: i64 = 0 73 o = cg_cat(out, o, "k:w" as *u8) 74 o = cg_catn(out, o, w) 75 o = cg_cat(out, o, "i" as *u8) 76 o = cg_catn(out, o, i) 77 out[o] = 0 as u8 78 return o 79} 80 81// one worker: write CAS_NPER distinct keys to `prefix`. locked=1 -> ws_put_locked else ws_put_p. 82func cas_worker(prefix: *u8, w: i64, locked: i64) -> i64 { 83 let key: *u8 = sys_mmap(64) 84 let val: *u8 = sys_mmap(64) 85 var i: i64 = 0 86 while i < CAS_NPER { 87 cg_key(key, w, i) 88 var o: i64 = 0 89 o = cg_cat(val, o, "v" as *u8) 90 o = cg_catn(val, o, w) 91 o = cg_cat(val, o, "_" as *u8) 92 o = cg_catn(val, o, i) 93 val[o] = 0 as u8 94 if locked == 1 { ws_put_locked(prefix, key, val) } 95 else { ws_put_p(prefix, key, val) } 96 i = i + 1 97 } 98 return 0 99} 100 101// fork CAS_NWORKERS workers over `prefix`, wait all. locked selects the write path. 102func cas_spawn(prefix: *u8, locked: i64) -> i64 { 103 let pids: *i64 = sys_mmap(8 * (CAS_NWORKERS + 4)) as *i64 104 var w: i64 = 0 105 while w < CAS_NWORKERS { 106 let pid: i64 = sys_fork() 107 if pid == 0 { 108 cas_worker(prefix, w, locked) 109 sys_exit(0) 110 } 111 pids[w] = pid 112 w = w + 1 113 } 114 let st: *i64 = sys_mmap(16) as *i64 115 w = 0 116 while w < CAS_NWORKERS { 117 sys_wait4(pids[w], st, 0) 118 w = w + 1 119 } 120 return 0 121} 122 123// count how many of the CAS_NWORKERS*CAS_NPER distinct keys are retrievable from `prefix` 124// (ss_open ONCE + ss_hget per key). Returns the retrievable count. 125func cas_retrievable(prefix: *u8) -> i64 { 126 let h: *i64 = ss_open(prefix) 127 let pq: *i64 = sys_mmap(16) as *i64 128 let lq: *i64 = sys_mmap(16) as *i64 129 let key: *u8 = sys_mmap(64) 130 var got: i64 = 0 131 var w: i64 = 0 132 while w < CAS_NWORKERS { 133 var i: i64 = 0 134 while i < CAS_NPER { 135 cg_key(key, w, i) 136 if ss_hget(h, key, pq, lq) == 1 { got = got + 1 } 137 i = i + 1 138 } 139 w = w + 1 140 } 141 return got 142} 143 144// build a per-run unique prefix into out: "/tmp/" + tag + <epoch> + "-" 145func cas_prefix(out: *u8, tag: *u8, epoch: i64) -> i64 { 146 var o: i64 = 0 147 o = cg_cat(out, o, "/tmp/" as *u8) 148 o = cg_cat(out, o, tag) 149 o = cg_catn(out, o, epoch) 150 o = cg_cat(out, o, "-" as *u8) 151 out[o] = 0 as u8 152 return o 153} 154 155func main() -> i64 { 156 let epoch: i64 = sys_now_realtime_sec() 157 let total: i64 = CAS_NWORKERS * CAS_NPER 158 159 let pfx_g: *u8 = sys_mmap(128) 160 let pfx_b: *u8 = sys_mmap(128) 161 cas_prefix(pfx_g, "casg" as *u8, epoch) 162 cas_prefix(pfx_b, "casb" as *u8, epoch) 163 164 // GOOD lane: concurrent LOCKED writes 165 cas_spawn(pfx_g, 1) 166 let good_got: i64 = cas_retrievable(pfx_g) 167 168 // BAD lane (NEG-CONTROL): concurrent UNLOCKED writes 169 cas_spawn(pfx_b, 0) 170 let bad_got: i64 = cas_retrievable(pfx_b) 171 172 var green: i64 = 1 173 if good_got != total { green = 0 } // locked must lose NOTHING 174 if bad_got >= total { green = 0 } // unlocked MUST lose something (neg-control) 175 176 so("WMS-R8 concurrent-SSOT CAS gate (flock-serialized registry write)\n" as *u8) 177 so(" good(locked) retrievable=" as *u8); son(good_got) 178 so(" want=" as *u8); son(total) 179 if good_got == total { so(" PASS\n" as *u8) } else { so(" FAIL(lost-update)\n" as *u8) } 180 so(" bad(unlocked) retrievable=" as *u8); son(bad_got) 181 so(" want<" as *u8); son(total) 182 if bad_got < total { so(" PASS(neg-control-sees-loss)\n" as *u8) } else { so(" FAIL(no-loss-detected)\n" as *u8) } 183 so("verdict=" as *u8) 184 if green == 1 { so("GREEN\n" as *u8) } else { so("RED\n" as *u8) } 185 186 // ONE assembled evidence record -> fa_appendz (lock-framed atomic self-log; dogfoods R0b) 187 let rec: *u8 = sys_mmap(CAS_REC_CAP + 16) 188 var o: i64 = 0 189 o = cg_cat(rec, o, "WMS-R8-CAS epoch=" as *u8); o = cg_catn(rec, o, epoch) 190 o = cg_cat(rec, o, " workers=" as *u8); o = cg_catn(rec, o, CAS_NWORKERS) 191 o = cg_cat(rec, o, " per=" as *u8); o = cg_catn(rec, o, CAS_NPER) 192 o = cg_cat(rec, o, " good=" as *u8); o = cg_catn(rec, o, good_got) 193 o = cg_cat(rec, o, " bad=" as *u8); o = cg_catn(rec, o, bad_got) 194 o = cg_cat(rec, o, " lost_bad=" as *u8); o = cg_catn(rec, o, total - bad_got) 195 o = cg_cat(rec, o, " verdict=" as *u8) 196 if green == 1 { o = cg_cat(rec, o, "GREEN" as *u8) } else { o = cg_cat(rec, o, "RED" as *u8) } 197 rec[o] = 0 as u8 198 fa_appendz(CAS_LOG, rec, CAS_REC_CAP) 199 200 if green == 1 { return 0 } 201 return 1 202}