nx_conductor_notes_migration_gate.nx source

1// nx_conductor_notes_migration_gate.nx -- WMS-R0b DELIVERABLE B referee. 2// 3// Proves the MIGRATED conductor-notes record write (multi-_fp/_fn sequence -> single 4// atomic locked fa_appendz) produces ZERO torn lines under concurrency, AND proves the 5// detector can actually SEE tearing via the mandatory in-gate NEGATIVE CONTROL: the OLD 6// _fp/_fn multi-write path, which MUST tear (bad_torn > 0) -- if it doesn't, the test is 7// worthless => RED (no false green). 8// 9// GOOD lane (the migration): N workers fork concurrently, each emits NRECS NOTES 10// records the NEW way -- assemble the whole "NOTES epoch=<e> weakest_arc=<arc> 11// call=...:triage ... verdict=DRILL-CALLED" line into one buffer + ONE 12// fa_appendz (the exact cn_emit_red shape). Expect ZERO torn lines and ALL 13// N*NRECS lines present. 14// BAD lane (NEGATIVE CONTROL): same concurrency, but each record emitted the ORIGINAL 15// way -- a SEQUENCE of separate _fp/_fn sys_write() calls to one O_APPEND fd 16// ("NOTES epoch=" then the number then " weakest_arc=" ...). Concurrent 17// workers interleave between those writes -> torn lines MUST appear. 18// TAMPER: an oversized record (rec_len+1 > cap) must be REJECTED with -2. 19// 20// A line is WELL-FORMED iff it starts "NOTES epoch=" AND its last 12 bytes before the '\n' 21// are "DRILL-CALLED" AND it contains exactly one "NOTES epoch=" head and exactly one 22// "DRILL-CALLED" tail. Any interleaving breaks one of those -> torn. Output: rows -> 23// stdout + knowledge/status/conductor_notes_migration_gate.log, then a verdict line. 24// Exit 0 on GREEN, 1 on RED. Sovereign: only nx_syscalls + nx_framed_append. 25// license_tier: ORIGINAL 26import "nx_syscalls.nx" 27import "nx_framed_append.nx" 28 29const NWORKERS: i64 = 16 // concurrency (matches framed_append_gate) 30const NRECS: i64 = 300 // records per worker 31const RECCAP: i64 = 256 // bounded record size (matches CN_REC_CAP) 32 33const GOODP: *u8 = "/tmp/cn_good.log\x00" as *u8 34const BADP: *u8 = "/tmp/cn_bad.log\x00" as *u8 35 36// dual-sink writer: stdout (fd 1) AND the evidence log fd 37func gp(logfd: i64, s: *u8) -> i64 { 38 var n: i64 = 0 39 while s[n] != (0 as u8) { n = n + 1 } 40 sys_write(1, s, n) 41 if logfd > 0 { sys_write(logfd, s, n) } 42 return 0 43} 44func gn(logfd: i64, v: i64) -> i64 { 45 let bb: *u8 = sys_mmap(28) 46 var m: i64 = v 47 if m < 0 { sys_write(1, "-\x00" as *u8, 1); if logfd > 0 { sys_write(logfd, "-\x00" as *u8, 1) } m = 0 - m } 48 let t: *u8 = sys_mmap(28) 49 var k: i64 = 0 50 if m == 0 { t[0] = 48 as u8; k = 1 } 51 while m > 0 { t[k] = (48 + (m % 10)) as u8; m = m / 10; k = k + 1 } 52 var i: i64 = 0 53 while i < k { bb[i] = t[k - 1 - i]; i = i + 1 } 54 sys_write(1, bb, k) 55 if logfd > 0 { sys_write(logfd, bb, k) } 56 return 0 57} 58 59// GOOD worker: NRECS migrated records via the SINGLE locked atomic fa_appendz path -- 60// the exact shape cn_emit_red writes. arc=<a number stand-in> keeps it deterministic. 61func good_worker(wid: i64) -> i64 { 62 let pid: i64 = __syscall(39, 0, 0, 0, 0, 0, 0) // getpid 63 let buf: *u8 = sys_mmap(RECCAP + 16) 64 var i: i64 = 0 65 while i < NRECS { 66 var o: i64 = 0 67 o = fa_cat(buf, o, "NOTES epoch=\x00" as *u8) 68 o = fa_catn(buf, o, pid) 69 o = fa_cat(buf, o, " weakest_arc=\x00" as *u8) 70 o = fa_catn(buf, o, wid) 71 o = fa_cat(buf, o, " call=sectional:triage reason=regression-chain-owns-red-arcs verdict=DRILL-CALLED\x00" as *u8) 72 fa_appendz(GOODP, buf, RECCAP) 73 i = i + 1 74 } 75 return 0 76} 77 78// inline decimal-into-buf helper for the BAD lane (mirrors the original _fn). 79func bad_num(fd: i64, v: i64) -> i64 { 80 let nb: *u8 = sys_mmap(28) 81 var m: i64 = v 82 var l: i64 = 0 83 if m == 0 { nb[0] = 48 as u8; l = 1 } 84 while m > 0 { nb[l] = (48 + (m % 10)) as u8; m = m / 10; l = l + 1 } 85 let d: *u8 = sys_mmap(28) 86 var x: i64 = 0 87 while x < l { d[x] = nb[l - 1 - x]; x = x + 1 } 88 sys_write(fd, d, l) 89 return 0 90} 91 92// BAD worker (NEGATIVE CONTROL): the ORIGINAL nx_conductor_notes write -- a SEQUENCE of 93// separate _fp/_fn sys_write() calls per NOTES record to one shared O_APPEND fd. Exactly 94// the torn-line bug the migration removes: concurrent workers interleave between writes. 95func bad_worker(wid: i64) -> i64 { 96 let pid: i64 = __syscall(39, 0, 0, 0, 0, 0, 0) // getpid 97 let fd: i64 = sys_openat_append(BADP, 0x1a4) 98 if fd < 0 { return 0 - 1 } 99 var i: i64 = 0 100 while i < NRECS { 101 sys_write(fd, "NOTES epoch=\x00" as *u8, 12) 102 bad_num(fd, pid) 103 sys_write(fd, " weakest_arc=\x00" as *u8, 13) 104 bad_num(fd, wid) 105 sys_write(fd, " call=sectional:triage reason=regression-chain-owns-red-arcs verdict=DRILL-CALLED\x00" as *u8, 81) 106 sys_write(fd, "\n\x00" as *u8, 1) 107 i = i + 1 108 } 109 sys_close(fd) 110 return 0 111} 112 113func spawn_all(which: i64) -> i64 { 114 let pids: *i64 = sys_mmap(8 * (NWORKERS + 4)) as *i64 115 var w: i64 = 0 116 while w < NWORKERS { 117 let pid: i64 = sys_fork() 118 if pid == 0 { 119 if which == 0 { good_worker(w) } 120 if which == 1 { bad_worker(w) } 121 sys_exit(0) 122 } 123 pids[w] = pid 124 w = w + 1 125 } 126 let st: *i64 = sys_mmap(16) as *i64 127 w = 0 128 while w < NWORKERS { 129 sys_wait4(pids[w], st, 0) 130 w = w + 1 131 } 132 return 0 133} 134 135// substring count of `pat` within [a,b) of hay. 136func slice_count(hay: *u8, a: i64, b: i64, pat: *u8, pl: i64) -> i64 { 137 var c: i64 = 0 138 var i: i64 = a 139 while i + pl <= b { 140 var k: i64 = 0 141 var hit: i64 = 1 142 while k < pl { if hay[i+k] != pat[k] { hit = 0; k = pl } else { k = k + 1 } } 143 if hit == 1 { c = c + 1 } 144 i = i + 1 145 } 146 return c 147} 148 149// parse a file: total_lines into outs[0], torn_count returned. 150// WELL-FORMED iff: begins "NOTES epoch=" AND the 12 bytes before '\n' are "DRILL-CALLED" 151// AND exactly one "NOTES epoch=" head AND exactly one "DRILL-CALLED" tail in the line. 152func count_torn(path: *u8, outs: *i64) -> i64 { 153 let szp: *i64 = sys_mmap(16) as *i64 154 let b: *u8 = sys_read_file(path, szp) 155 let sz: i64 = szp[0] 156 var torn: i64 = 0 157 var lines: i64 = 0 158 var ls: i64 = 0 159 var i: i64 = 0 160 let head: *u8 = "NOTES epoch=\x00" as *u8 // 12 bytes 161 let tail: *u8 = "DRILL-CALLED\x00" as *u8 // 12 bytes 162 while i < sz { 163 if b[i] == (10 as u8) { 164 let llen: i64 = i - ls 165 lines = lines + 1 166 var ok: i64 = 1 167 // prefix "NOTES epoch=" (12 bytes) 168 if llen < 24 { ok = 0 } 169 if ok == 1 { 170 var p: i64 = 0 171 while p < 12 { if b[ls + p] != head[p] { ok = 0 } p = p + 1 } 172 } 173 // suffix "DRILL-CALLED" immediately before '\n' 174 if ok == 1 { 175 let e: i64 = i - 12 176 var q: i64 = 0 177 while q < 12 { if b[e + q] != tail[q] { ok = 0 } q = q + 1 } 178 } 179 // exactly one head AND one tail inside the line (no interleaved 2nd record) 180 if ok == 1 { 181 if slice_count(b, ls, i, head, 12) != 1 { ok = 0 } 182 if slice_count(b, ls, i, tail, 12) != 1 { ok = 0 } 183 } 184 if ok == 0 { torn = torn + 1 } 185 ls = i + 1 186 } 187 i = i + 1 188 } 189 outs[0] = lines 190 return torn 191} 192 193func main() -> i64 { 194 let logfd: i64 = sys_openat_append("knowledge/status/conductor_notes_migration_gate.log\x00" as *u8, 0x1a4) 195 gp(logfd, "CN-MIGRATION epoch=\x00" as *u8) 196 gn(logfd, sys_now_realtime_sec()) 197 gp(logfd, " workers=\x00" as *u8); gn(logfd, NWORKERS) 198 gp(logfd, " recs=\x00" as *u8); gn(logfd, NRECS) 199 gp(logfd, " cap=\x00" as *u8); gn(logfd, RECCAP) 200 gp(logfd, "\n\x00" as *u8) 201 202 // fresh files each run (truncate) 203 let gtr: i64 = sys_openat_wr(GOODP, 0x1a4) 204 if gtr > 0 { sys_close(gtr) } 205 let btr: i64 = sys_openat_wr(BADP, 0x1a4) 206 if btr > 0 { sys_close(btr) } 207 208 // GOOD lane: concurrent MIGRATED single-write fa_appendz records 209 spawn_all(0) 210 // BAD lane (NEGATIVE CONTROL): same concurrency, ORIGINAL _fp/_fn multi-write 211 spawn_all(1) 212 213 let outs: *i64 = sys_mmap(16) as *i64 214 let good_torn: i64 = count_torn(GOODP, outs) 215 let good_lines: i64 = outs[0] 216 let bad_torn: i64 = count_torn(BADP, outs) 217 let bad_lines: i64 = outs[0] 218 let want_lines: i64 = NWORKERS * NRECS 219 220 // TAMPER: oversized record (rec_len+1 > cap=8) must be REJECTED with -2 221 let trec: *u8 = sys_mmap(64) 222 var ti: i64 = 0 223 while ti < 32 { trec[ti] = 65 as u8; ti = ti + 1 } 224 let tamper_rc: i64 = fa_append("/tmp/cn_tamper.log\x00" as *u8, trec, 32, 8) 225 226 gp(logfd, "CN row=good_lines got=\x00" as *u8); gn(logfd, good_lines) 227 gp(logfd, " want=\x00" as *u8); gn(logfd, want_lines) 228 if good_lines == want_lines { gp(logfd, " verdict=PASS\n\x00" as *u8) } else { gp(logfd, " verdict=FAIL\n\x00" as *u8) } 229 230 gp(logfd, "CN row=good_torn got=\x00" as *u8); gn(logfd, good_torn) 231 gp(logfd, " want=0\x00" as *u8) 232 if good_torn == 0 { gp(logfd, " verdict=PASS\n\x00" as *u8) } else { gp(logfd, " verdict=FAIL\n\x00" as *u8) } 233 234 gp(logfd, "CN row=neg_control bad_lines=\x00" as *u8); gn(logfd, bad_lines) 235 gp(logfd, " bad_torn=\x00" as *u8); gn(logfd, bad_torn) 236 gp(logfd, " want=POSITIVE\x00" as *u8) 237 if bad_torn > 0 { gp(logfd, " verdict=PASS(detector-sees-tearing)\n\x00" as *u8) } else { gp(logfd, " verdict=FAIL(no-tearing-detected)\n\x00" as *u8) } 238 239 gp(logfd, "CN row=tamper_oversize got=\x00" as *u8); gn(logfd, tamper_rc) 240 gp(logfd, " want=-2\x00" as *u8) 241 if tamper_rc == (0 - 2) { gp(logfd, " verdict=PASS\n\x00" as *u8) } else { gp(logfd, " verdict=FAIL\n\x00" as *u8) } 242 243 var green: i64 = 1 244 if good_lines != want_lines { green = 0 } 245 if good_torn != 0 { green = 0 } 246 if bad_torn <= 0 { green = 0 } 247 if tamper_rc != (0 - 2) { green = 0 } 248 249 gp(logfd, "CN-MIGRATION verdict=\x00" as *u8) 250 if green == 1 { gp(logfd, "GREEN\x00" as *u8) } else { gp(logfd, "RED\x00" as *u8) } 251 gp(logfd, " good_lines=\x00" as *u8); gn(logfd, good_lines) 252 gp(logfd, " good_torn=\x00" as *u8); gn(logfd, good_torn) 253 gp(logfd, " bad_torn=\x00" as *u8); gn(logfd, bad_torn) 254 gp(logfd, " tamper=\x00" as *u8); gn(logfd, tamper_rc) 255 if green == 0 { 256 gp(logfd, " reason=\x00" as *u8) 257 if good_lines != want_lines { gp(logfd, "good-lines-lost \x00" as *u8) } 258 if good_torn != 0 { gp(logfd, "GOOD-TORE \x00" as *u8) } 259 if bad_torn <= 0 { gp(logfd, "neg-control-saw-no-tearing \x00" as *u8) } 260 if tamper_rc != (0 - 2) { gp(logfd, "oversize-not-rejected \x00" as *u8) } 261 } 262 gp(logfd, "\n\x00" as *u8) 263 if logfd > 0 { sys_close(logfd) } 264 if green == 1 { return 0 } 265 return 1 266}