nx_statuscensus_lib.nx source
↩ module page · 335 lines · 16859 B
1// nx_statuscensus_lib.nx -- TS11 (/compare/trafficsafety): STATUS ANALYTICS + THE NON-EMPTY-200 SLA GAUGE, the LIB.
2//
3// OPERATOR 2026-09-05: "analytics on our 404s and 503s and all other statuses including undefined, and we
4// need to make sure we are getting to the point we have 99 percent and above 200 code slas with confirmed
5// non empty expected delivery." This lib is the ONE classifier; nx_statuscensus (program) runs it over the
6// live journals and nx_statuscensus_gate drives it over fixtures IN-PROCESS, so the gate and the beat cannot
7// disagree on a bucket. It EXTENDS the two journals the estate already writes -- no second request log:
8// EDGE sites_telemetry.log (nx_sites_telemetry.nx, one row per connection, append-only, 489 MB on
9// 2026-09-05, so the census reads a DECLARED TAIL WINDOW, never the whole file):
10// ts=<us> acc=<L|N> vh=<id> hs=<us> by=<bytes> rc=<0|-N>
11// vh=71 is the edge's own synthesized 503/504 (backend refused, edge window overrun, outcome-unknown,
12// deadline), vh=22 is the edge 404, rc<0 is a TLS handshake that never reached HTTP: there IS no
13// status for that row and it is counted UNDEFINED, never folded into a code. vh=70 is a PROXIED
14// backend reply: the edge row carries its bytes but NOT its status, so it counts DELIVERED-BY-BYTES
15// and is ALSO counted on a separate axis edge_proxied_unstatused= (the size of the blind spot); the
16// owed rung is a st=<code> field on the emitter so a proxied 500 with a body stops reading as a 200.
17// TOOLS knowledge/status/actlog.jrnl (ta_actlog_x): <ts> TAB mcp TAB <tool> TAB call TAB <status> TAB
18// tools/call lane=<l> exit=<n> bytes=<n> dur_ms=<n>. lane=async status=started rows are RECEIPTS
19// (the job's own result lands in _jobs/); a row with exit=0 and bytes>0 is DELIVERED; bytes=0 is EMPTY;
20// exit!=0 is ERROR even when bytes>0 (an error body is not a delivery).
21// PARTITIONS SUM (printed as edge_partition_sum= / act_partition_sum=), UNDEFINED is its own bucket, a window
22// under min_rows reports UNOBSERVED and does not vote, and the SLA permil is DELIVERED over ALL rows -- an empty
23// 200 is not a delivery. Verdict LAST line: GREEN rc 0 / RED rc 1 / UNOBSERVED rc 3.
24// license_tier: ORIGINAL No hw writes (Rule 26).
25import "nx_syscalls.nx"
26
27const SC_CONF_PATH: *u8 = "knowledge/statuscensus.conf"
28// 8 MiB of the edge log is ~200k rows at ~40 B/row: about a day of front-door traffic at the 2026-09-05 rate,
29// sized from the measured row width, not picked. The conf row window_bytes= overrides it.
30const SC_WINDOW_DEFAULT: i64 = 8388608
31const SC_SLA_MIN_DEFAULT: i64 = 990 // the operator's bar: 99 percent and above
32const SC_MINROWS_DEFAULT: i64 = 100 // below this a window is UNOBSERVED, never a pass
33const SC_VH_SYNTH: i64 = 71 // nx_sites_daemon_v2: edge-synthesized 503/504
34const SC_VH_404: i64 = 22 // nx_sites_daemon_v2: edge 404
35const SC_VH_PROXIED: i64 = 70 // nx_sites_daemon_v2: proxied backend reply, status not journaled
36const SC_SEEK_SET: i64 = 0
37const SC_SEEK_END: i64 = 2
38const SC_MODE_0644: i64 = 420
39const SC_NUMBUF: i64 = 32
40const SC_TAB: i64 = 9
41const SC_NL: i64 = 10
42const SC_MINUS: i64 = 45
43const SC_HASH: i64 = 35
44const SC_ZERO: i64 = 48
45const SC_NINE: i64 = 57
46
47func sc_slen(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } return n }
48func sc_fputs(fd: i64, s: *u8) -> i64 { sys_write(fd, s, sc_slen(s)); return 0 }
49func sc_fputn(fd: i64, v: i64) -> i64 {
50 let t: *u8 = sys_mmap(SC_NUMBUF)
51 let o: *u8 = sys_mmap(SC_NUMBUF)
52 var m: i64 = v
53 if m < 0 { sys_write(fd, "-" as *u8, 1); m = 0 - m }
54 var k: i64 = 0
55 if m == 0 { t[0] = SC_ZERO as u8; k = 1 }
56 while m > 0 { t[k] = (SC_ZERO + (m % 10)) as u8; m = m / 10; k = k + 1 }
57 var i: i64 = 0
58 while i < k { o[i] = t[k - 1 - i]; i = i + 1 }
59 sys_write(fd, o, k)
60 return 0
61}
62func sc_puts(s: *u8) -> i64 { return sc_fputs(1, s) }
63func sc_putn(v: i64) -> i64 { return sc_fputn(1, v) }
64
65// index of the newline ending the line that starts at i, or n (flag-driven: the cursor is never clobbered)
66func sc_eol(buf: *u8, i: i64, n: i64) -> i64 {
67 var e: i64 = i
68 var go: i64 = 1
69 while go == 1 { if e >= n { go = 0 } else { if buf[e] == (SC_NL as u8) { go = 0 } else { e = e + 1 } } }
70 return e
71}
72// substring key inside buf[s,e)
73func sc_has(buf: *u8, s: i64, e: i64, key: *u8) -> i64 {
74 let kl: i64 = sc_slen(key)
75 var i: i64 = s
76 while i + kl <= e {
77 var k: i64 = 0
78 var hit: i64 = 1
79 while k < kl { if buf[i + k] != key[k] { hit = 0; k = kl } else { k = k + 1 } }
80 if hit == 1 { return 1 }
81 i = i + 1
82 }
83 return 0
84}
85// occurrences of key inside buf[s,e) (non-overlapping)
86func sc_count(buf: *u8, s: i64, e: i64, key: *u8) -> i64 {
87 let kl: i64 = sc_slen(key)
88 var c: i64 = 0
89 var i: i64 = s
90 while i + kl <= e {
91 var k: i64 = 0
92 var hit: i64 = 1
93 while k < kl { if buf[i + k] != key[k] { hit = 0; k = kl } else { k = k + 1 } }
94 if hit == 1 { c = c + 1; i = i + kl } else { i = i + 1 }
95 }
96 return c
97}
98// decimal (optionally negative) right after the FIRST key inside buf[s,e); 1 found (out[0]=value) / 0 absent
99func sc_find_num(buf: *u8, s: i64, e: i64, key: *u8, out: *i64) -> i64 {
100 let kl: i64 = sc_slen(key)
101 var i: i64 = s
102 while i + kl <= e {
103 var k: i64 = 0
104 var hit: i64 = 1
105 while k < kl { if buf[i + k] != key[k] { hit = 0; k = kl } else { k = k + 1 } }
106 if hit == 1 {
107 var j: i64 = i + kl
108 var neg: i64 = 0
109 if j < e { if buf[j] == (SC_MINUS as u8) { neg = 1; j = j + 1 } }
110 var v: i64 = 0
111 var any: i64 = 0
112 var go: i64 = 1
113 while go == 1 {
114 if j >= e { go = 0 } else {
115 let c: i64 = buf[j] as i64
116 if c >= SC_ZERO { if c <= SC_NINE { v = v * 10 + (c - SC_ZERO); any = 1; j = j + 1 } else { go = 0 } } else { go = 0 }
117 }
118 }
119 if any == 0 { return 0 }
120 if neg == 1 { v = 0 - v }
121 out[0] = v
122 return 1
123 }
124 i = i + 1
125 }
126 return 0
127}
128// TAB s TAB, built at runtime (the lexer owns no tab escape and a literal tab in source is invisible)
129func sc_tabwrap(s: *u8) -> *u8 {
130 let n: i64 = sc_slen(s)
131 let b: *u8 = sys_mmap(n + 3)
132 b[0] = SC_TAB as u8
133 var i: i64 = 0
134 while i < n { b[1 + i] = s[i]; i = i + 1 }
135 b[n + 1] = SC_TAB as u8
136 b[n + 2] = 0 as u8
137 return b
138}
139
140// The TAIL window of an append-only file: open, size via lseek END, seek to size-window, read. lenbox[0] = bytes
141// read (or -1 when the file could not be opened: an ABSENT journal is UNOBSERVED, never zero rows), lenbox[1] =
142// the offset of the first COMPLETE row inside the window (a window that starts mid-line skips the cut row).
143func sc_read_tail(path: *u8, window: i64, lenbox: *i64) -> *u8 {
144 lenbox[0] = 0 - 1
145 lenbox[1] = 0
146 let fd: i64 = sys_openat_rd(path)
147 if fd < 0 { return 0 as *u8 }
148 let size: i64 = sys_lseek(fd, 0, SC_SEEK_END)
149 if size < 0 { sys_close(fd); return 0 as *u8 }
150 var start: i64 = size - window
151 if start < 0 { start = 0 }
152 let want: i64 = size - start
153 let buf: *u8 = sys_mmap(want + 1)
154 sys_lseek(fd, start, SC_SEEK_SET)
155 var got: i64 = 0
156 var go: i64 = 1
157 while go == 1 {
158 if got >= want { go = 0 } else {
159 let r: i64 = sys_read(fd, ((buf as i64) + got) as *u8, want - got)
160 if r <= 0 { go = 0 } else { got = got + r }
161 }
162 }
163 sys_close(fd)
164 var s: i64 = 0
165 if start > 0 { s = sc_eol(buf, 0, got); if s < got { s = s + 1 } }
166 lenbox[0] = got
167 lenbox[1] = s
168 return buf
169}
170
171// conf: window_bytes= sla_permil_min= min_rows= (absent file or row = the announced default; srcbox[0]=1 when conf)
172func sc_conf(key: *u8, dflt: i64, srcbox: *i64) -> i64 {
173 srcbox[0] = 0
174 let lb: *i64 = sys_mmap(16) as *i64
175 let buf: *u8 = sys_read_file(SC_CONF_PATH, lb)
176 if (buf as i64) == 0 { return dflt }
177 let n: i64 = lb[0]
178 let ob: *i64 = sys_mmap(16) as *i64
179 var i: i64 = 0
180 while i < n {
181 let e: i64 = sc_eol(buf, i, n)
182 if e > i { if buf[i] != (SC_HASH as u8) {
183 if sc_find_num(buf, i, e, key, ob) == 1 { srcbox[0] = 1; return ob[0] }
184 } }
185 i = e + 1
186 }
187 return dflt
188}
189
190// ---- counter slots: a partition each (the printer asserts the sum) + one separate axis ----
191const E_TOTAL: i64 = 0
192const E_DELIV: i64 = 1
193const E_EMPTY: i64 = 2
194const E_404: i64 = 3
195const E_SYNTH: i64 = 4
196const E_UNDEF: i64 = 5
197const E_PROXIED: i64 = 6 // SEPARATE AXIS, not a partition member (a proxied row is also DELIVERED or EMPTY)
198const E_SLOTS: i64 = 8
199
200func sc_census_edge(buf: *u8, s0: i64, n: i64, c: *i64) -> i64 {
201 let vb: *i64 = sys_mmap(16) as *i64
202 let bb: *i64 = sys_mmap(16) as *i64
203 let rb: *i64 = sys_mmap(16) as *i64
204 var i: i64 = s0
205 while i < n {
206 let e: i64 = sc_eol(buf, i, n)
207 if e > i { if sc_has(buf, i, e, "ts=" as *u8) == 1 { if sc_has(buf, i, e, " vh=" as *u8) == 1 {
208 c[E_TOTAL] = c[E_TOTAL] + 1
209 var vh: i64 = 0 - 1
210 var by: i64 = 0
211 var rc: i64 = 0
212 if sc_find_num(buf, i, e, " vh=" as *u8, vb) == 1 { vh = vb[0] }
213 if sc_find_num(buf, i, e, " by=" as *u8, bb) == 1 { by = bb[0] }
214 if sc_find_num(buf, i, e, " rc=" as *u8, rb) == 1 { rc = rb[0] }
215 if vh == SC_VH_PROXIED { c[E_PROXIED] = c[E_PROXIED] + 1 }
216 // ORDER IS THE PARTITION: a handshake that never reached HTTP has no status (UNDEFINED) even when the
217 // edge later wrote bytes; then the edge's own codes; then delivered-vs-empty by body bytes.
218 if rc < 0 { c[E_UNDEF] = c[E_UNDEF] + 1 }
219 else { if vh == SC_VH_SYNTH { c[E_SYNTH] = c[E_SYNTH] + 1 }
220 else { if vh == SC_VH_404 { c[E_404] = c[E_404] + 1 }
221 else { if by > 0 { c[E_DELIV] = c[E_DELIV] + 1 } else { c[E_EMPTY] = c[E_EMPTY] + 1 } } } }
222 } } }
223 i = e + 1
224 }
225 return c[E_TOTAL]
226}
227
228const A_TOTAL: i64 = 0
229const A_DELIV: i64 = 1
230const A_EMPTY: i64 = 2
231const A_ERROR: i64 = 3
232const A_RECEIPT: i64 = 4
233const A_SLOTS: i64 = 8
234
235func sc_census_act(buf: *u8, s0: i64, n: i64, c: *i64) -> i64 {
236 let xb: *i64 = sys_mmap(16) as *i64
237 let bb: *i64 = sys_mmap(16) as *i64
238 let started: *u8 = sc_tabwrap("started" as *u8)
239 var i: i64 = s0
240 while i < n {
241 let e: i64 = sc_eol(buf, i, n)
242 if e > i { if sc_has(buf, i, e, "tools/call lane=" as *u8) == 1 {
243 c[A_TOTAL] = c[A_TOTAL] + 1
244 var ec: i64 = 0
245 var by: i64 = 0
246 if sc_find_num(buf, i, e, " exit=" as *u8, xb) == 1 { ec = xb[0] }
247 if sc_find_num(buf, i, e, " bytes=" as *u8, bb) == 1 { by = bb[0] }
248 if sc_has(buf, i, e, started) == 1 { c[A_RECEIPT] = c[A_RECEIPT] + 1 }
249 else { if ec != 0 { c[A_ERROR] = c[A_ERROR] + 1 }
250 else { if by > 0 { c[A_DELIV] = c[A_DELIV] + 1 } else { c[A_EMPTY] = c[A_EMPTY] + 1 } } }
251 } }
252 i = e + 1
253 }
254 return c[A_TOTAL]
255}
256
257func sc_permil(num: i64, den: i64) -> i64 { if den <= 0 { return 0 - 1 } return (num * 1000) / den }
258
259// the same report to stdout and to the status fd, so the level and the console agree by construction
260func sc_report(fd: i64, ec: *i64, ac: *i64, window: i64, wsrc: i64, slamin: i64, minrows: i64, elen: i64, alen: i64, everdict: *u8, averdict: *u8, epm: i64, apm: i64, verdict: *u8) -> i64 {
261 sc_fputs(fd, "ts=" as *u8); sc_fputn(fd, sys_now_realtime_sec()); sc_fputs(fd, "\n" as *u8)
262 sc_fputs(fd, "window_bytes=" as *u8); sc_fputn(fd, window); if wsrc == 1 { sc_fputs(fd, " src=conf" as *u8) } else { sc_fputs(fd, " src=default" as *u8) }
263 sc_fputs(fd, " sla_permil_min=" as *u8); sc_fputn(fd, slamin); sc_fputs(fd, " min_rows=" as *u8); sc_fputn(fd, minrows); sc_fputs(fd, "\n" as *u8)
264 sc_fputs(fd, "edge_bytes_read=" as *u8); sc_fputn(fd, elen)
265 sc_fputs(fd, " edge_total=" as *u8); sc_fputn(fd, ec[E_TOTAL])
266 sc_fputs(fd, " edge_delivered=" as *u8); sc_fputn(fd, ec[E_DELIV])
267 sc_fputs(fd, " edge_empty=" as *u8); sc_fputn(fd, ec[E_EMPTY])
268 sc_fputs(fd, " edge_404=" as *u8); sc_fputn(fd, ec[E_404])
269 sc_fputs(fd, " edge_synth_5xx=" as *u8); sc_fputn(fd, ec[E_SYNTH])
270 sc_fputs(fd, " edge_undefined=" as *u8); sc_fputn(fd, ec[E_UNDEF])
271 sc_fputs(fd, " edge_partition_sum=" as *u8); sc_fputn(fd, ec[E_DELIV] + ec[E_EMPTY] + ec[E_404] + ec[E_SYNTH] + ec[E_UNDEF])
272 sc_fputs(fd, " edge_proxied_unstatused=" as *u8); sc_fputn(fd, ec[E_PROXIED])
273 sc_fputs(fd, " edge_sla_permil=" as *u8); sc_fputn(fd, epm); sc_fputs(fd, " edge=" as *u8); sc_fputs(fd, everdict); sc_fputs(fd, "\n" as *u8)
274 sc_fputs(fd, "act_bytes_read=" as *u8); sc_fputn(fd, alen)
275 sc_fputs(fd, " act_total=" as *u8); sc_fputn(fd, ac[A_TOTAL])
276 sc_fputs(fd, " act_delivered=" as *u8); sc_fputn(fd, ac[A_DELIV])
277 sc_fputs(fd, " act_empty=" as *u8); sc_fputn(fd, ac[A_EMPTY])
278 sc_fputs(fd, " act_error=" as *u8); sc_fputn(fd, ac[A_ERROR])
279 sc_fputs(fd, " act_receipts=" as *u8); sc_fputn(fd, ac[A_RECEIPT])
280 sc_fputs(fd, " act_partition_sum=" as *u8); sc_fputn(fd, ac[A_DELIV] + ac[A_EMPTY] + ac[A_ERROR] + ac[A_RECEIPT])
281 sc_fputs(fd, " act_sla_permil=" as *u8); sc_fputn(fd, apm); sc_fputs(fd, " act=" as *u8); sc_fputs(fd, averdict); sc_fputs(fd, "\n" as *u8)
282 sc_fputs(fd, "verdict=" as *u8); sc_fputs(fd, verdict); sc_fputs(fd, "\n" as *u8)
283 return 0
284}
285
286// the census over two journals with explicit bars (so a gate drives it on fixtures); 0 GREEN / 1 RED / 3 UNOBSERVED
287func sc_run(telpath: *u8, actpath: *u8, statuspath: *u8, tmppath: *u8, logpath: *u8, window: i64, wsrc: i64, slamin: i64, minrows: i64) -> i64 {
288 let elb: *i64 = sys_mmap(32) as *i64
289 let alb: *i64 = sys_mmap(32) as *i64
290 let ebuf: *u8 = sc_read_tail(telpath, window, elb)
291 let abuf: *u8 = sc_read_tail(actpath, window, alb)
292 let ec: *i64 = sys_mmap(E_SLOTS * 8) as *i64
293 let ac: *i64 = sys_mmap(A_SLOTS * 8) as *i64
294 if elb[0] > 0 { sc_census_edge(ebuf, elb[1], elb[0], ec) }
295 if alb[0] > 0 { sc_census_act(abuf, alb[1], alb[0], ac) }
296 var epm: i64 = 0 - 1
297 var apm: i64 = 0 - 1
298 var everdict: *u8 = "UNOBSERVED" as *u8
299 var averdict: *u8 = "UNOBSERVED" as *u8
300 var red: i64 = 0
301 var voted: i64 = 0
302 if ec[E_TOTAL] >= minrows {
303 epm = sc_permil(ec[E_DELIV], ec[E_TOTAL]); voted = voted + 1
304 if epm >= slamin { everdict = "GREEN" as *u8 } else { everdict = "RED" as *u8; red = 1 }
305 }
306 if ac[A_TOTAL] >= minrows {
307 apm = sc_permil(ac[A_DELIV], ac[A_TOTAL]); voted = voted + 1
308 if apm >= slamin { averdict = "GREEN" as *u8 } else { averdict = "RED" as *u8; red = 1 }
309 }
310 var verdict: *u8 = "UNOBSERVED" as *u8
311 var rc: i64 = 3
312 if voted > 0 { if red == 1 { verdict = "RED" as *u8; rc = 1 } else { verdict = "GREEN" as *u8; rc = 0 } }
313 sc_puts("=== NX-STATUSCENSUS -- every response status the estate returned, UNDEFINED its own bucket, delivery = a body AND no error ===\n" as *u8)
314 sc_report(1, ec, ac, window, wsrc, slamin, minrows, elb[0], alb[0], everdict, averdict, epm, apm, verdict)
315 // the LEVEL: truncate-written via tmp+rename so a reader never sees a torn file
316 let sfd: i64 = sys_openat_wr(tmppath, SC_MODE_0644)
317 if sfd >= 0 {
318 sc_report(sfd, ec, ac, window, wsrc, slamin, minrows, elb[0], alb[0], everdict, averdict, epm, apm, verdict)
319 sys_close(sfd)
320 sys_renameat(tmppath, statuspath)
321 }
322 // the TRAJECTORY: one row per run, append-only, so improving or regressing is readable
323 let lfd: i64 = sys_openat_append(logpath, SC_MODE_0644)
324 if lfd >= 0 {
325 sc_fputs(lfd, "ts=" as *u8); sc_fputn(lfd, sys_now_realtime_sec())
326 sc_fputs(lfd, " edge_total=" as *u8); sc_fputn(lfd, ec[E_TOTAL]); sc_fputs(lfd, " edge_delivered=" as *u8); sc_fputn(lfd, ec[E_DELIV])
327 sc_fputs(lfd, " edge_undefined=" as *u8); sc_fputn(lfd, ec[E_UNDEF]); sc_fputs(lfd, " edge_synth_5xx=" as *u8); sc_fputn(lfd, ec[E_SYNTH]); sc_fputs(lfd, " edge_404=" as *u8); sc_fputn(lfd, ec[E_404])
328 sc_fputs(lfd, " edge_proxied_unstatused=" as *u8); sc_fputn(lfd, ec[E_PROXIED])
329 sc_fputs(lfd, " edge_sla_permil=" as *u8); sc_fputn(lfd, epm)
330 sc_fputs(lfd, " act_total=" as *u8); sc_fputn(lfd, ac[A_TOTAL]); sc_fputs(lfd, " act_delivered=" as *u8); sc_fputn(lfd, ac[A_DELIV]); sc_fputs(lfd, " act_sla_permil=" as *u8); sc_fputn(lfd, apm)
331 sc_fputs(lfd, " verdict=" as *u8); sc_fputs(lfd, verdict); sc_fputs(lfd, "\n" as *u8)
332 sys_close(lfd)
333 }
334 return rc
335}