nx_pub_submit_gate.nx source

1// nx_pub_submit_gate.nx -- R1 proof for THE NISHI PUBLISHER intake: under a REAL multi-process race (PG_N 2// sys_fork children x PG_K pub_submit calls each = the many-workstreams-publish-at-once scenario), the durable 3// framed queue keeps EVERY request and tears NONE -- while the UNLOCKED neg-control (same record, multi-write, 4// no flock) DOES tear under identical load. That is the exact collision that crashed sites.elf, measured 5// before/after. Sovereign: our fork/wait4 + our framed-append + our scan. Per-wsid test queue (nx_runpath) so 6// sibling workstreams' gate runs don't contend. GREEN iff framed = all-present & zero-torn AND unlocked tears. 7// license_tier: ORIGINAL expect_exit: 0 8import "nx_syscalls.nx" 9import "nx_publisher.nx" 10import "nx_runpath.nx" 11 12const PG_N: i64 = 4 13const PG_K: i64 = 100 14 15func g_w(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} sys_write(1,s,n); return 0 } 16func g_n(v: i64) -> i64 { var m: i64=v; if m<0{g_w("-" as *u8);m=0-m} let t:*u8=sys_mmap(24); 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; let o:*u8=sys_mmap(24); while i<k{o[i]=t[k-1-i];i=i+1} sys_write(1,o,k); return 0 } 17func g_row(id: *u8, ok: i64, pass: *i64) -> i64 { g_w(" " as *u8); g_w(id); g_w(": " as *u8); if ok==1 { g_w("OK\n" as *u8); pass[0]=pass[0]+1 } else { g_w("FAIL\n" as *u8) } return 0 } 18 19// build dest "nf/g<child>_<k>.html" (unique per request -> proves no record was lost, not just counted). 20func pg_dest(child: i64, k: i64, out: *u8) -> i64 { 21 var o: i64 = 0 22 o = fa_cat(out, o, "nf/g" as *u8); o = fa_catn(out, o, child); o = fa_cat(out, o, "_" as *u8); o = fa_catn(out, o, k); o = fa_cat(out, o, ".html" as *u8) 23 out[o] = 0 as u8 24 return o 25} 26func pg_make_src(srcpath: *u8) -> i64 { 27 let fd: i64 = sys_openat_wr(srcpath, 0x1a4) 28 if fd < 0 { return 0 } 29 sys_write(fd, "nishi publisher r1 test artifact\n" as *u8, 33) 30 sys_close(fd) 31 return 1 32} 33func pg_reset(qpath: *u8) -> i64 { let fd: i64 = sys_openat_wr(qpath, 0x1a4); if fd >= 0 { sys_close(fd) } return 0 } 34 35// scan the queue: out[0]=well-formed line count, out[1]=torn lines (a record has EXACTLY 6 tabs / 7 fields). 36func pg_scan(qpath: *u8, out: *i64) -> i64 { 37 out[0] = 0; out[1] = 0 38 let lenp: *i64 = sys_mmap(8) as *i64 39 let data: *u8 = sys_read_file(qpath, lenp) 40 if (data as i64) == 0 { return 0 } 41 let n: i64 = lenp[0] 42 var i: i64 = 0 43 var tabs: i64 = 0 44 var lines: i64 = 0 45 var torn: i64 = 0 46 while i < n { 47 let c: i64 = data[i] & 0xff 48 if c == 9 { tabs = tabs + 1 } 49 if c == 10 { lines = lines + 1; if tabs != 6 { torn = torn + 1 } tabs = 0 } 50 i = i + 1 51 } 52 out[0] = lines; out[1] = torn 53 return 0 54} 55 56func pg_child(qpath: *u8, srcpath: *u8, child: i64, good: i64) -> i64 { 57 var k: i64 = 0 58 while k < PG_K { 59 let dest: *u8 = sys_mmap(64) 60 pg_dest(child, k, dest) 61 if good == 1 { pub_submit_to(qpath, srcpath, dest, "nishifamily" as *u8, "ws_test" as *u8, "internal" as *u8) } 62 else { pub_append_unlocked(qpath, srcpath, dest, "nishifamily" as *u8, "ws_test" as *u8, "internal" as *u8) } 63 k = k + 1 64 } 65 sys_exit(0) 66 return 0 67} 68func pg_phase(qpath: *u8, srcpath: *u8, good: i64, out: *i64) -> i64 { 69 pg_reset(qpath) 70 var i: i64 = 0 71 while i < PG_N { let pid: i64 = sys_fork(); if pid == 0 { pg_child(qpath, srcpath, i, good) } i = i + 1 } 72 let status: *i64 = sys_mmap(8) as *i64 73 i = 0 74 while i < PG_N { sys_wait4(0 - 1, status, 0); i = i + 1 } 75 pg_scan(qpath, out) 76 return 0 77} 78 79func main() -> i64 { 80 let pass: *i64 = sys_mmap(8) as *i64; pass[0]=0 81 g_w("=== NX-PUBLISHER SUBMIT GATE (durable concurrency-safe queue: N forks x K submits) ===\n" as *u8) 82 let wsid: *u8 = sys_mmap(64); rp_wsid(wsid); rp_ensure(wsid) 83 let qpath: *u8 = sys_mmap(512); rp_path(wsid, "pub_q_test" as *u8, qpath) 84 let srcpath: *u8 = sys_mmap(512); rp_path(wsid, "pub_src_test" as *u8, srcpath) 85 pg_make_src(srcpath) 86 87 let expected: i64 = PG_N * PG_K 88 let go: *i64 = sys_mmap(16) as *i64 89 let bo: *i64 = sys_mmap(16) as *i64 90 pg_phase(qpath, srcpath, 1, go) // FRAMED (pub_submit_to -> fa_appendz) 91 pg_phase(qpath, srcpath, 0, bo) // UNLOCKED neg-control (multi-write, no flock) 92 93 g_w(" config: "); g_n(PG_N); g_w(" submitters x "); g_n(PG_K); g_w(" requests expected="); g_n(expected); g_w("\n") 94 g_w(" FRAMED (pub_submit): lines="); g_n(go[0]); g_w(" torn="); g_n(go[1]); g_w("\n") 95 g_w(" UNLOCKED neg-control: lines="); g_n(bo[0]); g_w(" torn="); g_n(bo[1]); g_w("\n") 96 97 g_row("DURABLE: every request landed (framed lines == expected, no loss)" as *u8, (go[0]==expected) as i64, pass) 98 g_row("INTACT: zero torn records under concurrency (framed torn == 0)" as *u8, (go[1]==0) as i64, pass) 99 g_row("RACE IS REAL: the unlocked path tears under identical load (neg-control)" as *u8, (bo[1]>0) as i64, pass) 100 g_row("MEASURED F->S: framed strictly fewer torn than unlocked (the before/after)" as *u8, (go[1] < bo[1]) as i64, pass) 101 102 g_w("PUB-SUBMIT rows=4 pass="); g_n(pass[0]) 103 if pass[0]==4 { g_w(" verdict=GREEN\n"); sys_exit(0); return 0 } 104 g_w(" verdict=RED\n"); sys_exit(1); return 1 105}