nx_replyreserve.nx source
↩ module page · 257 lines · 13638 B
1// nx_replyreserve.nx -- CLI for the reply-reserve calibrator (lane E, 2026-08-25). Thin main; the ruler is
2// nx_replyreserve_lib.rr_derive and the quantile machinery is nx_loadceil_lib, so nx_replyreserve_gate
3// composes both in-process on planted buffers.
4//
5// nx_replyreserve [report|apply|beat] [logpath]
6// beat (DEFAULT) Derive and print, then apply ONLY if replyreserve-auto-apply is 1 in the conf.
7// report READ-ONLY. Derive and print; never touches the conf.
8// apply Explicit operator action: rewrite knowledge/edge_window.conf when a ratchet is due.
9//
10// WHY THE DEFAULT IS THE WRITING VERB, deliberately and against least-surprise: the clocksched plane's
11// organ column is a BARE PATH -- every live row is `<name> <interval> <elf>` with no argv -- so the clock
12// forks this organ with no arguments. If the bare invocation were the read-only one, the beat would be
13// structurally inert and this organ could never arm itself no matter how long it ran, which is the whole
14// point of building it. Safety does not come from which verb is default; it comes from the conf switch
15// (replyreserve-auto-apply), the bounded envelope [floor_ms, cap_ms], the refusal that keeps the current
16// value, and the journal line every application writes. `report` is one word away for a read-only look.
17//
18// exit 0 RATCHET(-ready or applied) / 1 HOLD (conf already carries the measured value) / 2 UNREADABLE /
19// 3 INSUFFICIENT-HISTORY (abstain) / 4 REFUSED-OUT-OF-BOUNDS (current value KEPT).
20// HOLD and RATCHET are both successes; they are distinct codes so a beat can tell "changed" from "already
21// right", and INSUFFICIENT is distinct from UNREADABLE so a blind axis never reads as a healthy one.
22//
23// THE LOG IS READ FROM ITS TAIL, NOT ITS HEAD, and that is load-bearing rather than an optimisation:
24// actlog.jrnl is append-only and 30 MB, reply_ms= has only existed since the 2026-08-23 build, so every
25// sample lives in the last few KB. A capped read from offset 0 returns 4 MB of history that predates the
26// field entirely and yields ZERO samples -- an abstention for the wrong reason, from an organ that could
27// then never arm itself no matter how long it ran. Coverage is printed either way.
28// license_tier: ORIGINAL
29
30import "nx_replyreserve_lib.nx"
31
32const RR_OUTCAP: i64 = 32768
33
34func rrw(fd: i64, s: *u8) -> i64 { let n: i64 = lc_slen(s); if n > 0 { sys_write(fd, s, n) } return 0 }
35func rro(s: *u8) -> i64 { return rrw(1, s) }
36func rrn(v0: i64) -> i64 {
37 let b: *u8 = sys_mmap(32)
38 let e: i64 = rr_catn(b, 0, v0)
39 b[e] = 0 as u8
40 sys_write(1, b, e)
41 sys_munmap(b, 32)
42 return 0
43}
44func rr_word(v: i64) -> *u8 {
45 if v == RR_RATCHET { return "RATCHET" as *u8 }
46 if v == RR_HOLD { return "HOLD" as *u8 }
47 if v == RR_UNREADABLE { return "UNREADABLE" as *u8 }
48 if v == RR_INSUFFICIENT { return "INSUFFICIENT-HISTORY" as *u8 }
49 return "REFUSED-OUT-OF-BOUNDS" as *u8
50}
51
52// truncate-write a status artifact with a canonical LAST line, so a consumer anchors by POSITION and a
53// stale verdict can never be re-read as a fresh one (the gate-roster law: the producer owes a rewritten log).
54func rr_status_write(out: *i64, verdict: i64, applied: i64, epoch: i64, whole: i64, size: i64) -> i64 {
55 let b: *u8 = sys_mmap(4096)
56 var p: i64 = rr_cat(b, 0, "NX-REPLYRESERVE ts=" as *u8)
57 p = rr_catn(b, p, epoch)
58 p = rr_cat(b, p, " samples=" as *u8); p = rr_catn(b, p, out[RR_O_N])
59 p = rr_cat(b, p, " need=" as *u8); p = rr_catn(b, p, out[RR_O_MINN])
60 p = rr_cat(b, p, " lines_scanned=" as *u8); p = rr_catn(b, p, out[RR_O_LINES])
61 p = rr_cat(b, p, " window_whole_file=" as *u8); p = rr_catn(b, p, whole)
62 p = rr_cat(b, p, " log_bytes=" as *u8); p = rr_catn(b, p, size)
63 p = rr_catc(b, p, LC_NL)
64 p = rr_cat(b, p, "reply_ms min=" as *u8); p = rr_catn(b, p, out[RR_O_MIN])
65 p = rr_cat(b, p, " p50=" as *u8); p = rr_catn(b, p, out[RR_O_P50])
66 p = rr_cat(b, p, " p90=" as *u8); p = rr_catn(b, p, out[RR_O_P90])
67 p = rr_cat(b, p, " p99=" as *u8); p = rr_catn(b, p, out[RR_O_P99])
68 p = rr_cat(b, p, " max=" as *u8); p = rr_catn(b, p, out[RR_O_MAX])
69 p = rr_catc(b, p, LC_NL)
70 p = rr_cat(b, p, "current_ms=" as *u8); p = rr_catn(b, p, out[RR_O_CURRENT])
71 p = rr_cat(b, p, " derived_ms=" as *u8); p = rr_catn(b, p, out[RR_O_DERIVED])
72 p = rr_cat(b, p, " window_ms=" as *u8); p = rr_catn(b, p, out[RR_O_WINDOW])
73 p = rr_cat(b, p, " cap_ms=" as *u8); p = rr_catn(b, p, out[RR_O_CAPMS])
74 p = rr_cat(b, p, " floor_ms=" as *u8); p = rr_catn(b, p, out[RR_O_FLOOR])
75 p = rr_cat(b, p, " applied=" as *u8); p = rr_catn(b, p, applied)
76 p = rr_catc(b, p, LC_NL)
77 p = rr_cat(b, p, "verdict=" as *u8)
78 p = rr_cat(b, p, rr_word(verdict))
79 p = rr_catc(b, p, LC_NL)
80 let fd: i64 = sys_openat_wr(RR_STATUS, MODE_0644)
81 if fd >= 0 { sys_write(fd, b, p); sys_close(fd) }
82 sys_munmap(b, 4096)
83 return 0
84}
85
86func rr_jrnl_append(oldv: i64, newv: i64, n: i64, epoch: i64) -> i64 {
87 let b: *u8 = sys_mmap(512)
88 var p: i64 = rr_cat(b, 0, "REPLYRESERVE-APPLIED ts=" as *u8)
89 p = rr_catn(b, p, epoch)
90 p = rr_cat(b, p, " from_ms=" as *u8); p = rr_catn(b, p, oldv)
91 p = rr_cat(b, p, " to_ms=" as *u8); p = rr_catn(b, p, newv)
92 p = rr_cat(b, p, " samples=" as *u8); p = rr_catn(b, p, n)
93 p = rr_catc(b, p, LC_NL)
94 let fd: i64 = sys_openat_append(RR_JRNL, MODE_0644)
95 if fd >= 0 { sys_write(fd, b, p); sys_close(fd) }
96 sys_munmap(b, 512)
97 return 0
98}
99
100// Atomic conf swap: write <conf>.rrnew, fsync, rename over the live path. The serving daemon reads this
101// conf on EVERY sync call, so a plain truncate-write would expose a torn read on the request path.
102func rr_publish(path: *u8, tmp: *u8, b: *u8, n: i64) -> i64 {
103 let fd: i64 = sys_openat_wr(tmp, MODE_0644)
104 if fd < 0 { return 0 - 1 }
105 let w: i64 = sys_write(fd, b, n)
106 sys_fsync(fd)
107 sys_close(fd)
108 if w != n { return 0 - 1 }
109 return sys_renameat(tmp, path)
110}
111
112func main(argc: i64, argv: *i64) -> i64 {
113 // 0=report 1=apply 2=beat. Defaults to BEAT because the clock forks this organ with no argv; see
114 // the header note. An explicit `report` selects the read-only path.
115 var verb: i64 = 2
116 var a: i64 = 1
117 if argc >= 2 {
118 let op: *u8 = argv[1] as *u8
119 if op[0] == (97 as u8) { verb = 1 }
120 if op[0] == (98 as u8) { verb = 2 }
121 if op[0] == (114 as u8) { verb = 0 }
122 a = 2
123 }
124 var logp: *u8 = RR_LOG
125 if argc > a { logp = argv[a] as *u8 }
126
127 // ---- thresholds: conf rows, line-anchored, each with a NAMED fallback ----
128 let cb: *u8 = sys_mmap(LC_CONFCAP)
129 var cn: i64 = lc_read(RR_CONF, cb, LC_CONFCAP)
130 if cn < 0 { cn = 0 }
131 // p99, because the conf row this organ exists to replace asks in its own words for "the measured
132 // reply_ms p99". 200 samples matches the sibling calibrator's floor and, at the arrival rate observed
133 // on 2026-08-25 (8 promoted replies in 461 s, about one per minute), is roughly three hours of real
134 // traffic -- enough that the value is history rather than a coincidence. Observed rate is a rough
135 // BOUND from one window, not a rate guarantee.
136 let q: i64 = lc_conf_line_num(cb, cn, "replyreserve-quantile-permil" as *u8, 990)
137 let minn: i64 = lc_conf_line_num(cb, cn, "replyreserve-min-samples" as *u8, 200)
138 let margin: i64 = lc_conf_line_num(cb, cn, "replyreserve-margin-permil" as *u8, 2000)
139 let floorms: i64 = lc_conf_line_num(cb, cn, "replyreserve-floor-ms" as *u8, 250)
140 let maxpermil: i64 = lc_conf_line_num(cb, cn, "replyreserve-max-permil-of-window" as *u8, 100)
141 let tailb: i64 = lc_conf_line_num(cb, cn, "replyreserve-tail-bytes" as *u8, LC_LOGCAP)
142 let autoap: i64 = lc_conf_line_num(cb, cn, "replyreserve-auto-apply" as *u8, 1)
143
144 // ---- the consumer's own conf, read with the consumer's own parser contract ----
145 let wb: *u8 = sys_mmap(LC_CONFCAP)
146 let wn0: i64 = lc_read(RR_WINDOW_CONF, wb, LC_CONFCAP)
147 if wn0 <= 0 {
148 rro("NX-REPLYRESERVE REFUSED window-conf-unreadable path=" as *u8); rro(RR_WINDOW_CONF); rro("\n" as *u8)
149 rro("verdict=UNREADABLE\n" as *u8)
150 return RR_UNREADABLE
151 }
152 let window: i64 = lc_conf_line_num(wb, wn0, RR_WINDOW_KEY, 0)
153 let current: i64 = lc_conf_line_num(wb, wn0, RR_RESERVE_KEY, 0 - 1)
154 if window <= 0 {
155 rro("NX-REPLYRESERVE REFUSED no edge_window_ms row in " as *u8); rro(RR_WINDOW_CONF); rro("\n" as *u8)
156 rro("verdict=UNREADABLE\n" as *u8)
157 return RR_UNREADABLE
158 }
159
160 // ---- the samples, from the TAIL ----
161 let meta: *i64 = sys_mmap(LC_M_N * 8) as *i64
162 let buf: *u8 = sys_mmap(tailb)
163 let n: i64 = lc_read_tail(logp, buf, tailb, meta)
164 if n < 0 {
165 rro("NX-REPLYRESERVE REFUSED log-unreadable path=" as *u8); rro(logp); rro("\n" as *u8)
166 rro("verdict=UNREADABLE\n" as *u8)
167 return RR_UNREADABLE
168 }
169
170 let out: *i64 = sys_mmap(RR_O_SLOTS * 8) as *i64
171 let v: i64 = rr_derive(buf, n, q, minn, margin, floorms, maxpermil, window, current, out)
172 let epoch: i64 = sys_now_realtime_sec()
173
174 rro("NX-REPLYRESERVE log=" as *u8); rro(logp)
175 rro(" log_bytes=" as *u8); rrn(meta[LC_M_SIZE])
176 rro(" window_bytes=" as *u8); rrn(meta[LC_M_BYTES])
177 rro(" whole_file=" as *u8); rrn(meta[LC_M_WHOLE])
178 rro(" lines_scanned=" as *u8); rrn(out[RR_O_LINES])
179 rro(" without_reply_ms=" as *u8); rrn(out[RR_O_MISSING])
180 rro(" capped=" as *u8); rrn(out[RR_O_CAPPED]); rro("\n" as *u8)
181 rro(" subject: reply_ms= on lane=sync-promoted rows. samples=" as *u8); rrn(out[RR_O_N])
182 rro(" need>=" as *u8); rrn(out[RR_O_MINN]); rro("\n" as *u8)
183 if out[RR_O_N] > 0 {
184 rro(" reply_ms: min=" as *u8); rrn(out[RR_O_MIN])
185 rro(" p50=" as *u8); rrn(out[RR_O_P50])
186 rro(" p90=" as *u8); rrn(out[RR_O_P90])
187 rro(" p99=" as *u8); rrn(out[RR_O_P99])
188 rro(" max=" as *u8); rrn(out[RR_O_MAX]); rro("\n" as *u8)
189 }
190 rro(" bounds: window_ms=" as *u8); rrn(out[RR_O_WINDOW])
191 rro(" current_ms=" as *u8); rrn(out[RR_O_CURRENT])
192 rro(" floor_ms=" as *u8); rrn(out[RR_O_FLOOR])
193 rro(" cap_ms=" as *u8); rrn(out[RR_O_CAPMS])
194 rro(" margin_permil=" as *u8); rrn(out[RR_O_MARGIN]); rro("\n" as *u8)
195
196 var applied: i64 = 0
197 if v == RR_INSUFFICIENT {
198 rro(" ABSTAINING: " as *u8); rrn(out[RR_O_N])
199 rro(" samples is below the " as *u8); rrn(out[RR_O_MINN])
200 rro(" needed, so reply_reserve_ms KEEPS its current value of " as *u8); rrn(out[RR_O_CURRENT])
201 rro(" ms.\n This is the honest state, not a failure: the reply_ms field has only existed since the\n" as *u8)
202 rro(" 2026-08-23 build, so the history is still accruing. Nothing was written. Re-run when the\n" as *u8)
203 rro(" sample reaches the floor; the beat does this without a human returning.\n" as *u8)
204 }
205 if v == RR_REFUSED {
206 rro(" REFUSED: derived " as *u8); rrn(out[RR_O_DERIVED])
207 rro(" ms is outside the armed envelope (must be >0, < window " as *u8); rrn(out[RR_O_WINDOW])
208 rro(", and <= cap " as *u8); rrn(out[RR_O_CAPMS])
209 rro("). The current value " as *u8); rrn(out[RR_O_CURRENT])
210 rro(" ms is KEPT -- a calibrator refuses toward leaving things alone.\n" as *u8)
211 }
212 if v == RR_HOLD {
213 rro(" HOLD: the conf already carries the measured value " as *u8); rrn(out[RR_O_DERIVED])
214 rro(" ms. Nothing to do.\n" as *u8)
215 }
216 if v == RR_RATCHET {
217 rro(" quantile_value=" as *u8); rrn(out[RR_O_QVAL])
218 rro(" scaled_by_margin=" as *u8); rrn(out[RR_O_SCALED])
219 rro(" observed_max=" as *u8); rrn(out[RR_O_MAX])
220 rro(" -> DERIVED reply_reserve_ms=" as *u8); rrn(out[RR_O_DERIVED])
221 rro(" (was " as *u8); rrn(out[RR_O_CURRENT]); rro(")\n" as *u8)
222 var doit: i64 = 0
223 if verb == 1 { doit = 1 }
224 if verb == 2 { if autoap == 1 { doit = 1 } }
225 if doit == 1 {
226 let ob: *u8 = sys_mmap(RR_OUTCAP)
227 let nn: i64 = rr_rewrite(wb, wn0, ob, RR_OUTCAP, out[RR_O_DERIVED], out[RR_O_CURRENT],
228 out[RR_O_N], out[RR_O_QVAL], out[RR_O_Q], epoch)
229 if nn < 0 {
230 rro(" APPLY REFUSED: could not rewrite the conf (code " as *u8); rrn(nn)
231 rro("; -1 = no reply_reserve_ms row to replace, -2 = output buffer too small). Nothing written.\n" as *u8)
232 } else {
233 let pr: i64 = rr_publish(RR_WINDOW_CONF, "knowledge/edge_window.conf.rrnew" as *u8, ob, nn)
234 if pr == 0 {
235 applied = 1
236 rr_jrnl_append(out[RR_O_CURRENT], out[RR_O_DERIVED], out[RR_O_N], epoch)
237 rro(" APPLIED: " as *u8); rro(RR_WINDOW_CONF)
238 rro(" rewritten atomically, row and provenance paragraph together, now " as *u8); rrn(nn)
239 rro(" bytes.\n" as *u8)
240 } else {
241 rro(" APPLY FAILED: atomic publish returned " as *u8); rrn(pr)
242 rro(" -- the live conf is UNCHANGED.\n" as *u8)
243 }
244 }
245 } else {
246 if verb == 2 {
247 rro(" NOT APPLIED: replyreserve-auto-apply is 0, so the beat reports and leaves the conf alone.\n" as *u8)
248 } else {
249 rro(" report is READ-ONLY. Run `nx_replyreserve apply` to arm this value.\n" as *u8)
250 }
251 }
252 }
253
254 rr_status_write(out, v, applied, epoch, meta[LC_M_WHOLE], meta[LC_M_SIZE])
255 rro("verdict=" as *u8); rro(rr_word(v)); rro("\n" as *u8)
256 return v
257}