nx_pub_resilient_gate.nx source

1// nx_pub_resilient_gate.nx -- proves pub_run_resilient is UN-STUCKABLE (operator: "the publisher cant get stuck 2// again as a whole even if somethings get stuck"). The queue is [good1, CRASHER, good2]; the CRASHER has a 16 KB 3// dest field that overflows the child's fixed buffer and SIGSEGVs the child. Liar-killed proof: good2 -- the item 4// AFTER the crasher -- STILL lands live (head-of-line blocking eliminated), the crasher is QUARANTINED to the 5// dead-letter (reason=CRASH), and the drain reports published=2 crashed=1. In the old in-process pub_run_governed 6// this same queue would kill the whole process at the crasher and good2 would never publish. license_tier: ORIGINAL expect_exit: 0 7import "nx_publisher.nx" 8import "nx_syscalls.nx" 9 10func w(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} sys_write(1,s,n); return 0 } 11func wn(v: i64) -> i64 { var m: i64=v; if m<0{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 } 12func dmk(p: *u8) -> i64 { __syscall(83, p as i64, 493, 0, 0, 0, 0); return 0 } 13func dun(p: *u8) -> i64 { __syscall(87, p as i64, 0, 0, 0, 0, 0); return 0 } 14func mkfile(p: *u8, c: *u8) -> i64 { let fd: i64 = sys_openat_wr(p, 0x1a4); if fd<0 { return 0 } var n: i64=0; while c[n]!=(0 as u8){n=n+1} sys_write(fd,c,n); sys_close(fd); return 1 } 15func qcat(buf: *u8, o: i64, s: *u8) -> i64 { var w2: i64=o; var i: i64=0; while s[i]!=(0 as u8){buf[w2]=s[i];w2=w2+1;i=i+1} return w2 } 16func file_has(path: *u8, ndl: *u8) -> i64 { 17 let lp: *i64 = sys_mmap(8) as *i64; let d: *u8 = sys_read_file(path, lp); if (d as i64)==0 { return 0 } 18 let n: i64 = lp[0]; var nl: i64=0; while ndl[nl]!=(0 as u8){nl=nl+1} 19 var i: i64=0; while i+nl<=n { var j: i64=0; var ok: i64=1; while j<nl{ if d[i+j]!=ndl[j]{ok=0;j=nl} else {j=j+1} } if ok==1{return 1} i=i+1 } return 0 20} 21 22func main(argc: i64, argv: *i64) -> i64 { 23 w("=== nx_pub_resilient_gate -- the un-stuckable drain (crash in the middle) ===\n" as *u8) 24 dmk("/tmp/prg" as *u8); dmk("/tmp/prg/stage" as *u8); dmk("/tmp/prg/live" as *u8); dmk("/tmp/prg/appr" as *u8) 25 // reset state so the gate is rerunnable (ledger/dead-letter/live empty) 26 dun("/tmp/prg/led.tsv" as *u8); dun("/tmp/prg/dl.tsv" as *u8); dun("/tmp/prg/q.tsv" as *u8) 27 dun("/tmp/prg/live/page1.html" as *u8); dun("/tmp/prg/live/page3.html" as *u8) 28 mkfile("/tmp/prg/src.html" as *u8, "GOOD-PUBLISHED-CONTENT\n" as *u8) 29 let sha: *u8 = sys_mmap(72) 30 if pub_sha_file("/tmp/prg/src.html" as *u8, sha) == 0 { w(" cannot sha src\n" as *u8); sys_exit(1); return 1 } 31 let z: *u8 = "0000000000000000000000000000000000000000000000000000000000000000" as *u8 32 pub_record_ledger("/tmp/prg/led.tsv" as *u8, sha, "page4.html" as *u8, "site" as *u8) // src.html's CURRENT content is already live at page4.html (stale-dup setup) 33 34 // build queue: good1(page1), BAD(missing src -> publish fails in its child), good2(page3 -- AFTER the bad one) 35 let q: *u8 = sys_mmap(8192); var o: i64 = 0 36 o=qcat(q,o,"PENDING" as *u8);q[o]=9 as u8;o=o+1; o=qcat(q,o,"site" as *u8);q[o]=9 as u8;o=o+1; o=qcat(q,o,"ws" as *u8);q[o]=9 as u8;o=o+1; o=qcat(q,o,sha);q[o]=9 as u8;o=o+1; o=qcat(q,o,"/tmp/prg/src.html" as *u8);q[o]=9 as u8;o=o+1; o=qcat(q,o,"page1.html" as *u8);q[o]=9 as u8;o=o+1; o=qcat(q,o,"internal" as *u8);q[o]=10 as u8;o=o+1 37 // BAD: source does not exist -> the child's pub_publish_one fails (exit 2) WITHOUT taking down the drain 38 o=qcat(q,o,"PENDING" as *u8);q[o]=9 as u8;o=o+1; o=qcat(q,o,"site" as *u8);q[o]=9 as u8;o=o+1; o=qcat(q,o,"ws" as *u8);q[o]=9 as u8;o=o+1; o=qcat(q,o,sha);q[o]=9 as u8;o=o+1; o=qcat(q,o,"/tmp/prg/MISSING.html" as *u8);q[o]=9 as u8;o=o+1; o=qcat(q,o,"page2.html" as *u8);q[o]=9 as u8;o=o+1; o=qcat(q,o,"internal" as *u8);q[o]=10 as u8;o=o+1 39 o=qcat(q,o,"PENDING" as *u8);q[o]=9 as u8;o=o+1; o=qcat(q,o,"site" as *u8);q[o]=9 as u8;o=o+1; o=qcat(q,o,"ws" as *u8);q[o]=9 as u8;o=o+1; o=qcat(q,o,sha);q[o]=9 as u8;o=o+1; o=qcat(q,o,"/tmp/prg/src.html" as *u8);q[o]=9 as u8;o=o+1; o=qcat(q,o,"page3.html" as *u8);q[o]=9 as u8;o=o+1; o=qcat(q,o,"internal" as *u8);q[o]=10 as u8;o=o+1 40 // STALE DUP: queued sha is stale (zeros) but src.html's current content is already live at page4.html -> must SKIP, not fail/quarantine 41 o=qcat(q,o,"PENDING" as *u8);q[o]=9 as u8;o=o+1; o=qcat(q,o,"site" as *u8);q[o]=9 as u8;o=o+1; o=qcat(q,o,"ws" as *u8);q[o]=9 as u8;o=o+1; o=qcat(q,o,z);q[o]=9 as u8;o=o+1; o=qcat(q,o,"/tmp/prg/src.html" as *u8);q[o]=9 as u8;o=o+1; o=qcat(q,o,"page4.html" as *u8);q[o]=9 as u8;o=o+1; o=qcat(q,o,"internal" as *u8);q[o]=10 as u8;o=o+1 42 q[o]=0 as u8 43 let qfd: i64 = sys_openat_wr("/tmp/prg/q.tsv" as *u8, 0x1a4); sys_write(qfd, q, o); sys_close(qfd) 44 45 // DRAIN resiliently 46 let oc: *i64 = sys_mmap(64) as *i64 47 pub_run_resilient("/tmp/prg/q.tsv" as *u8, "/tmp/prg/led.tsv" as *u8, "/tmp/prg/stage" as *u8, "/tmp/prg/live" as *u8, "prg_lock" as *u8, "/tmp/prg/appr" as *u8, "/tmp/prg/dl.tsv" as *u8, oc) 48 w(" drain: published="); wn(oc[0]); w(" held="); wn(oc[1]); w(" failed="); wn(oc[2]); w(" crashed="); wn(oc[3]); w(" skipped="); wn(oc[4]); w("\n" as *u8) 49 50 var pass: i64 = 0; var tot: i64 = 0 51 // T1 the item AFTER the bad one published (HEAD-OF-LINE BLOCKING ELIMINATED) -- the core proof 52 tot=tot+1; if pub_exists("/tmp/prg/live/page3.html" as *u8)==1 { pass=pass+1; w(" PASS page3 (after the bad item) is LIVE -> drain did NOT halt\n" as *u8) } else { w(" FAIL page3 not live -> the bad item halted the drain\n" as *u8) } 53 // T2 the item BEFORE published too 54 tot=tot+1; if pub_exists("/tmp/prg/live/page1.html" as *u8)==1 { pass=pass+1; w(" PASS page1 is LIVE\n" as *u8) } else { w(" FAIL page1 not live\n" as *u8) } 55 // T3 the bad item did NOT publish (no half/ghost publish) 56 tot=tot+1; if pub_exists("/tmp/prg/live/page2.html" as *u8)==0 { pass=pass+1; w(" PASS page2 (bad item) NOT live\n" as *u8) } else { w(" FAIL page2 published anyway\n" as *u8) } 57 // T4 exactly 2 published; exactly 1 isolated as fail-or-crash 58 tot=tot+1; if oc[0]==2 { if (oc[2]+oc[3])==1 { pass=pass+1; w(" PASS counts: published=2, isolated(fail|crash)=1\n" as *u8) } else { w(" FAIL isolated count != 1\n" as *u8) } } else { w(" FAIL published != 2\n" as *u8) } 59 // T5 the bad item was QUARANTINED to the dead-letter (so the self-heal can root-cause it) 60 tot=tot+1; if file_has("/tmp/prg/dl.tsv" as *u8, "STUCK" as *u8)==1 { pass=pass+1; w(" PASS bad item quarantined to dead-letter\n" as *u8) } else { w(" FAIL bad item not dead-lettered\n" as *u8) } 61 // T6 DIRECT crash-isolation: a child that genuinely SIGSEGVs must NOT take down its parent (the crash branch) 62 let cpid: i64 = sys_fork() 63 if cpid == 0 { let bad: *u8 = 0 as *u8; bad[0] = 65 as u8; sys_exit(0) } // store to NULL -> SIGSEGV 64 let cst: *i64 = sys_mmap(16) as *i64; cst[0] = 0; sys_wait4(cpid, cst, 0) 65 let csig: i64 = cst[0] & 0x7f 66 tot=tot+1; if csig != 0 { pass=pass+1; w(" PASS parent SURVIVED a real SIGSEGV child (sig="); wn(csig); w(") -> crash branch isolates\n" as *u8) } else { w(" FAIL child did not crash as expected\n" as *u8) } 67 // T7 the STALE DUP (page4) was SKIPPED cleanly -- NOT failed, NOT quarantined (the drain self-recognizes already-live content -> no per-pass churn) 68 tot=tot+1; if oc[4] >= 1 { if file_has("/tmp/prg/dl.tsv" as *u8, "page4.html" as *u8)==0 { pass=pass+1; w(" PASS stale-dup SKIPPED cleanly (skipped="); wn(oc[4]); w(", not failed/quarantined)\n" as *u8) } else { w(" FAIL stale-dup was quarantined\n" as *u8) } } else { w(" FAIL stale-dup not skipped (skipped=0)\n" as *u8) } 69 70 w("nx_pub_resilient_gate pass="); wn(pass); w("/"); wn(tot) 71 if pass==tot { w(" verdict=GREEN (one crashing item CANNOT stall the publisher; it is isolated + quarantined)\n" as *u8); sys_exit(0); return 0 } 72 w(" verdict=RED\n" as *u8); sys_exit(1); return 1 73}