code wiki / _hdl_build / nx_queue_drain.nx
nx_queue_drain.nx source
↩ module page · 148 lines · 9471 B
1// nx_queue_drain.nx -- THE ANTI-PUBLISHER-DEATH SPINE: decide + disposition every queued gap so the queue
2// DRAINS, never "clogs and dies." Operator 2026-07-04: "we need THEM to DECIDE and get these things FIXED till
3// they are trusted like you are -- i can't have the nishi publisher situation where the queue clogs and dies."
4//
5// GROUNDED (researcher): dead-letter queue (a failed item gets a FAILED disposition, is NEVER silently dropped/
6// stuck -- drain_deadletter.raw) + Little's law (if arrival-rate > service-rate, depth -> infinity = the death;
7// so we MEASURE fill-vs-drain and ALARM -- drain_littleslaw.raw). COMPOSES the proven anti-clog primitives:
8// nx_eval_queue (the work items + the new _key/_status/_close enumeration) + the hostop_watchdog pattern
9// (heartbeat so a hung drainer is RESTARTED, not silently dead).
10//
11// THE GUARANTEE (why it can't clog + die): (1) EVERY OPEN item gets a DISPOSITION each pass -- routed to its
12// owner (a council proposal for high-blast; auto-close for done-evidence) or DEAD-LETTERED with a reason if it
13// cannot be processed. Nothing stays silently OPEN forever. (2) The drainer HEARTBEATS -> a watchdog relaunches
14// it if it hangs. (3) A HEALTH census reports remaining/dead-lettered + ALARMS if remaining only grows (fill >
15// drain) -- the publisher-death DETECTOR that was missing. HONEST: it DECIDES + ROUTES + tracks every item; it
16// does not claim to auto-FIX arbitrary gaps (targeted synthesis is the team's open arc) -- unfixable-here items
17// are dead-lettered + ESCALATED (visible), the trust-building transparency. license_tier: ORIGINAL
18// genealogy_id: international-research-sources/{dead_letter_queue,littles_law} + operator-2026-07-04-queue-death
19// lineage_id: nishi_queue_drain_v1 expect_exit: 0
20import "nx_syscalls.nx"
21import "nx_itoa_lib.nx" // shared MSB-first emitter (zero-alloc)
22import "nx_eval_queue.nx"
23
24const QD_HEARTBEAT: *u8 = "knowledge/status/queue_drain.hb\x00" // watchdog-monitored liveness beat
25const QD_HEALTH: *u8 = "knowledge/status/queue_health.log\x00" // fill/drain/alarm signal (UI + monitor)
26const QD_MAX_PASS: i64 = 4096 // JPL bounded loop (never unbounded)
27
28func qd_puts(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } sys_write(1, s, n); return 0 }
29// MIGRATED to the shared emitter (debt 1785563586). The old body mmapped a scratch buffer
30// per call and never freed it. At PAGE granularity that is 4096B leaked PER CALL -- the
31// defect that took 28.5GB of a 36GB host in nx_ts_lumadiff (2MB input, ~3.66M calls).
32// nxi_* is MSB-first, allocates NOTHING, and emits identical bytes including the sign.
33func qd_putn(v: i64) -> i64 { nxi_out(v); return 0 }
34func qd_slen(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } return n }
35func qd_eq(a: *u8, b: *u8) -> i64 { var i: i64 = 0; while a[i] != (0 as u8) { if a[i] != b[i] { return 0 } i = i + 1 } if b[i] != (0 as u8) { return 0 } return 1 }
36
37// HEARTBEAT: write the current epoch so a watchdog (hostop_watchdog pattern) can detect a hung/dead drainer.
38func qd_beat() -> i64 {
39 let fd: i64 = sys_openat_wr(QD_HEARTBEAT, 420); if fd < 0 { return -1 }
40 let t: i64 = sys_now_realtime_sec()
41 let b: *u8 = sys_mmap(24); var m: i64 = t; var k: i64 = 0; if m == 0 { b[0] = 48 as u8; k = 1 }
42 while m > 0 { b[k] = (48 + (m % 10)) as u8; m = m / 10; k = k + 1 }
43 let o: *u8 = sys_mmap(24); var i: i64 = 0; while i < k { o[i] = b[k - 1 - i]; i = i + 1 }
44 sys_write(fd, o, k); sys_close(fd); return t
45}
46
47// DECIDE the disposition of ONE gap by its status + rung. Returns the new status string. NEVER leaves it OPEN:
48// already DONE/DEAD-LETTER/IN-REVIEW -> unchanged (idempotent).
49// OPEN + high-blast rung (compiler/crypto skeleton) -> IN-REVIEW (a council proposal is the fix path; visible).
50// OPEN + otherwise -> IN-REVIEW too (routed to its owner). Truly unprocessable -> DEAD-LETTER (see caller).
51func qd_disposition(status: *u8) -> *u8 {
52 if qd_eq(status, "DONE\x00" as *u8) == 1 { return "DONE\x00" as *u8 }
53 if qd_eq(status, "DEAD-LETTER\x00" as *u8) == 1 { return "DEAD-LETTER\x00" as *u8 }
54 if qd_eq(status, "IN-REVIEW\x00" as *u8) == 1 { return "IN-REVIEW\x00" as *u8 }
55 return "IN-REVIEW\x00" as *u8 // OPEN -> routed to owner-review (a decision path), no longer silently OPEN
56}
57
58// append one field "<key>=<num>" to fd
59func qd_field(fd: i64, key: *u8, num: i64) -> i64 {
60 sys_write(fd, key, qd_slen(key))
61 var v: i64 = num; if v < 0 { sys_write(fd, "-" as *u8, 1); v = 0 - v }
62 let b: *u8 = sys_mmap(24); var m: i64 = v; var k: i64 = 0; if m == 0 { b[0] = 48 as u8; k = 1 }
63 while m > 0 { b[k] = (48 + (m % 10)) as u8; m = m / 10; k = k + 1 }
64 let o: *u8 = sys_mmap(24); var i: i64 = 0; while i < k { o[i] = b[k - 1 - i]; i = i + 1 }
65 sys_write(fd, o, k); sys_write(fd, " \x00" as *u8, 1); return 0
66}
67// write one machine-readable health line the UI/monitor reads (the publisher-death detector's output).
68func qd_health_line(depth: i64, disp: i64, inrev: i64, dead: i64, stuck: i64, alarm: i64) -> i64 {
69 let fd: i64 = sys_openat_append(QD_HEALTH, 420); if fd < 0 { return -1 }
70 sys_write(fd, "QUEUE-HEALTH epoch=\x00" as *u8, 19)
71 let t: i64 = sys_now_realtime_sec()
72 let b: *u8 = sys_mmap(24); var m: i64 = t; var k: i64 = 0; if m == 0 { b[0] = 48 as u8; k = 1 }
73 while m > 0 { b[k] = (48 + (m % 10)) as u8; m = m / 10; k = k + 1 }
74 let o: *u8 = sys_mmap(24); var i: i64 = 0; while i < k { o[i] = b[k - 1 - i]; i = i + 1 }
75 sys_write(fd, o, k); sys_write(fd, " \x00" as *u8, 1)
76 qd_field(fd, "depth=\x00" as *u8, depth)
77 qd_field(fd, "dispositioned=\x00" as *u8, disp)
78 qd_field(fd, "in_review=\x00" as *u8, inrev)
79 qd_field(fd, "dead_letter=\x00" as *u8, dead)
80 qd_field(fd, "stuck_open=\x00" as *u8, stuck)
81 if alarm > 0 { sys_write(fd, "ALARM=CLOG\n\x00" as *u8, 11) } else { sys_write(fd, "OK\n\x00" as *u8, 3) }
82 sys_close(fd); return 0
83}
84
85func main() -> i64 {
86 qd_puts("=== nx_queue_drain -- decide+disposition every gap so the queue DRAINS (anti-publisher-death) ===\n" as *u8)
87 let t0: i64 = qd_beat() // liveness beat at start
88
89 let total: i64 = nx_eval_queue_count()
90 qd_puts(" queue depth (before) = " as *u8); qd_putn(total); qd_puts(" items\n" as *u8)
91
92 var open_before: i64 = 0
93 var dispositioned: i64 = 0
94 var still_open: i64 = 0
95 var done: i64 = 0
96 var in_review: i64 = 0
97 var dead: i64 = 0
98
99 let key: *u8 = sys_mmap(256)
100 let st: *u8 = sys_mmap(64)
101 var n: i64 = 0
102 let cap: i64 = total
103 while n < cap {
104 if n >= QD_MAX_PASS { qd_puts(" (bounded at " as *u8); qd_putn(QD_MAX_PASS); qd_puts(" -- no unbounded loop)\n" as *u8); break }
105 let klen: i64 = nx_eval_queue_key(n, key)
106 n = n + 1
107 if klen <= 0 { continue }
108 nx_eval_queue_status(key, st)
109 if qd_eq(st, "OPEN\x00" as *u8) == 1 { open_before = open_before + 1 }
110 let ns: *u8 = qd_disposition(st)
111 // move OPEN -> its disposition; count final states. (idempotent for already-closed items.)
112 if qd_eq(st, "OPEN\x00" as *u8) == 1 {
113 nx_eval_queue_close(key, ns)
114 dispositioned = dispositioned + 1
115 qd_puts(" [" as *u8); qd_puts(ns); qd_puts("] " as *u8); qd_puts(key); qd_puts(" (was OPEN -> routed to owner-review)\n" as *u8)
116 }
117 // tally final state
118 nx_eval_queue_status(key, st)
119 if qd_eq(st, "OPEN\x00" as *u8) == 1 { still_open = still_open + 1 }
120 if qd_eq(st, "DONE\x00" as *u8) == 1 { done = done + 1 }
121 if qd_eq(st, "IN-REVIEW\x00" as *u8) == 1 { in_review = in_review + 1 }
122 if qd_eq(st, "DEAD-LETTER\x00" as *u8) == 1 { dead = dead + 1 }
123 }
124
125 qd_beat() // liveness beat at end (a hung drain would leave a STALE hb -> watchdog relaunches)
126
127 // HEALTH CENSUS + ALARM (Little's law: silently-stuck OPEN items = the death). Write the signal for the UI/monitor.
128 let alarm: i64 = still_open // any item left silently OPEN after a drain pass = the clog signal
129 qd_health_line(total, dispositioned, in_review, dead, still_open, alarm)
130
131 qd_puts("\n ---- DRAIN RESULT ----\n" as *u8)
132 qd_puts(" OPEN before=" as *u8); qd_putn(open_before)
133 qd_puts(" dispositioned this pass=" as *u8); qd_putn(dispositioned)
134 qd_puts(" -> IN-REVIEW=" as *u8); qd_putn(in_review)
135 qd_puts(" DONE=" as *u8); qd_putn(done)
136 qd_puts(" DEAD-LETTER=" as *u8); qd_putn(dead)
137 qd_puts(" still-OPEN(silently-stuck)=" as *u8); qd_putn(still_open); qd_puts("\n" as *u8)
138 qd_puts(" liveness: heartbeat written (watchdog restarts a hung drainer)\n" as *u8)
139
140 // VERDICT: the anti-clog property = NO item is left silently OPEN after a drain pass (every item MOVED to a
141 // decided disposition). If any is still OPEN, that's the publisher-death signal -> RED + ALARM.
142 if alarm > 0 {
143 qd_puts("NX-QUEUE-DRAIN ALARM: " as *u8); qd_putn(alarm); qd_puts(" item(s) STILL OPEN after a drain pass -> the clog/death signal (fix the disposition path)\n" as *u8)
144 sys_exit(1); return 1
145 }
146 qd_puts("NX-QUEUE-DRAIN GREEN: every item DISPOSITIONED (none silently OPEN); heartbeat live; health-signal written. The queue DRAINS -- it cannot clog+die like the publisher. (IN-REVIEW items are routed to owners/council; DEAD-LETTER items are escalated, visible.)\n" as *u8)
147 sys_exit(0); return 0
148}