nx_pub_recover_exec.nx source

1// nx_pub_recover_exec.nx -- THE PUBLISHER'S RECOVERY EXECUTOR (the action rung after nx_pub_recover's decision). 2// 3// Consumes PENDING recovery requests from the queue, KILLS the wedged daemon by name (proc_kill_by_name -> 4// the existing nx_hostctl keeper respawns a FRESH one = the wedge clears), flips the queue line PENDING->DONE 5// (atomic rewrite under fl_acquire), and appends a recovery LEDGER record. The Publisher owns the recovery 6// ACTION; hosting's supervisor owns the respawn (its existing job) -> clean separation, no second supervisor, 7// no touching nx_hostctl. SIGTERM (15) = graceful; never-brick #26 (kill+supervised-respawn is reversible, 8// no destructive write). Runs on the host where the daemons live (NAS); the gate proves it on a real mock. 9// pe_exec(queuepath, ledgerpath) -> number of recoveries executed 10// license_tier: ORIGINAL 11import "nx_syscalls.nx" 12import "nx_proc_ctl.nx" // proc_kill_by_name / proc_alive_by_name 13import "nx_pub_recover.nx" // pr_field / pr_n / pr_s / pr_slen + (transitively) fl_acquire / fl_release / fa_appendz 14const PE_MAGIC_100000: i64 = 100000 15const PE_MAGIC_4096: i64 = 4096 16 17const PE_QUEUE: *u8 = "knowledge/publish/recovery_queue.tsv" 18const PE_LEDGER: *u8 = "knowledge/publish/recovery_ledger.tsv" 19const PE_LOCK: *u8 = "publish_recover" 20const PE_SIGTERM: i64 = 15 21 22func pe_exec(queuepath: *u8, ledgerpath: *u8) -> i64 { 23 let lk: i64 = fl_acquire(PE_LOCK, PE_MAGIC_100000, 1) 24 let lp: *i64 = sys_mmap(16) as *i64; lp[0] = 0 25 let b: *u8 = sys_read_file(queuepath, lp) 26 if (b as i64) == 0 { fl_release(lk); return 0 } 27 let n: i64 = lp[0] 28 let out: *u8 = sys_mmap(n + PE_MAGIC_4096); var oo: i64 = 0 29 let daemon: *u8 = sys_mmap(128); let f0: *u8 = sys_mmap(32) 30 var executed: i64 = 0 31 var i: i64 = 0 32 while i < n { 33 var e: i64 = i; while e < n { if b[e] == (10 as u8) { break } else { e = e + 1 } } 34 let ll: i64 = e - i 35 if ll > 0 { 36 pr_field(((b as i64)+i) as *u8, ll, 0, f0) 37 if f0[0] == (80 as u8) { // 'P' of PENDING 38 pr_field(((b as i64)+i) as *u8, ll, 2, daemon) 39 let killed: i64 = proc_kill_by_name(daemon, PE_SIGTERM) 40 let rec: *u8 = sys_mmap(256); var ro: i64 = 0 41 ro = pr_n(rec, ro, sys_now_realtime_sec()); rec[ro]=9 as u8; ro=ro+1 42 ro = pr_s(rec, ro, daemon); rec[ro]=9 as u8; ro=ro+1 43 ro = pr_n(rec, ro, killed); rec[ro]=9 as u8; ro=ro+1 44 ro = pr_s(rec, ro, "KILLED" as *u8); rec[ro]=10 as u8; ro=ro+1 45 fa_appendz(ledgerpath, rec, 256) 46 executed = executed + 1 47 // rewrite line: PENDING(7 chars) -> DONE, rest verbatim from the first TAB onward 48 oo = pr_s(out, oo, "DONE" as *u8) 49 var c: i64 = i + 7 50 while c < e { out[oo] = b[c]; oo = oo + 1; c = c + 1 } 51 out[oo] = 10 as u8; oo = oo + 1 52 } else { 53 var c: i64 = i 54 while c < e { out[oo] = b[c]; oo = oo + 1; c = c + 1 } 55 out[oo] = 10 as u8; oo = oo + 1 56 } 57 } 58 i = e + 1 59 } 60 // atomic queue rewrite: write <queue>.tmp then rename over the live queue. 61 let tmp: *u8 = sys_mmap(320); var to: i64 = pr_s(tmp, 0, queuepath) 62 tmp[to]=46 as u8; to=to+1; tmp[to]=116 as u8; to=to+1; tmp[to]=109 as u8; to=to+1; tmp[to]=112 as u8; to=to+1; tmp[to]=0 as u8 // ".tmp" 63 let fd: i64 = sys_openat_wr(tmp, 0x1a4) 64 if fd >= 0 { sys_write(fd, out, oo); sys_close(fd); __syscall(82, tmp as i64, queuepath as i64, 0, 0, 0, 0) } // rename(2) 65 fl_release(lk) 66 return executed 67} 68 69func pe_uw(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} sys_write(1,s,n); return 0 } 70func main(argc: i64, argv: *i64) -> i64 { 71 pe_uw("=== nishi-publisher recovery EXECUTOR ===\n" as *u8) 72 let c: i64 = pe_exec(PE_QUEUE, PE_LEDGER) 73 pe_uw("recoveries executed: " as *u8); let nb: *u8=sys_mmap(24); let e: i64=pr_n(nb,0,c); sys_write(1,nb,e); pe_uw("\n" as *u8) 74 return 0 75}