nx_cronwatch_put.nx source
↩ module page · 293 lines · 16113 B
1// nx_cronwatch_put.nx -- THE MISSING PRODUCER for the cron dead-man's-switch registry (2026-08-15).
2//
3// WHY THIS EXISTS. nx_cron_watch is the estate's dead-man's switch: every scheduled job pings by
4// leaving a `ts=<epoch>` in its own evidence log, and a check-in missed beyond grace reports STALE.
5// Its manifest moved to the seg-store plane knowledge/store/cronwatch- -- and NOTHING COULD WRITE IT.
6// nx_store_put projects its own row shape onto a foreign grammar (the exact hazard nx_clockjob was
7// built for on the clock plane, where a half-written row still sits in the live registry with no name
8// at all, skipped by the dispatcher on every window).
9// A RECONCILER WITH NO PRODUCER IS A CONSUMER WITH NOTHING TO CONSUME, and THE ABSENCE OF A
10// SANCTIONED WRITER DOES NOT STOP WRITES -- IT ONLY MAKES THEM MALFORMED.
11//
12// nx_cronwatch_put put <name> <heartbeat-file> <ts-marker> <max-age-secs> <remediation>
13// nx_cronwatch_put selftest
14//
15// Grammar is EXACTLY nx_cron_watch's: name<TAB>heartbeat-file<TAB>ts-marker<TAB>max-age<TAB>remediation.
16// Validated at the boundary so a malformed row never reaches the plane, because a consumer can only
17// SKIP a bad row after the fact.
18//
19// MAX-AGE IS NOT A NUMBER TO GUESS. It is expected-interval + grace, and the caller derives it from
20// the interval it registered on the clock. interval + interval/4 is the recommended shape: it absorbs
21// scheduler jitter while still catching the FIRST missed fire, because a single miss puts age at two
22// intervals which is already past the bound. This organ does not invent it -- it refuses a
23// non-positive one and records what it was given.
24//
25// TWO GUARDS THIS ORGAN HAS THAT ITS CLOCK-PLANE SIBLING DOES NOT, both earned from the same law --
26// A DETECTOR WITH FALSE POSITIVES IS WORSE THAN NONE, BECAUSE IT TEACHES EVERYONE TO IGNORE IT:
27// 1. the heartbeat file must EXIST, and
28// 2. the ts-marker must be FINDABLE IN IT RIGHT NOW.
29// A watch declared over a missing file, or over a marker that never appears, reports STALE from the
30// moment it is written and never stops. That is not a dead-man's switch, it is a stuck alarm, and it
31// is worse than no watch at all because it trains the reader to skip the whole board.
32//
33// ONE SUBJECT PER ROW: a duplicate name is refused. Two rows for one job means the check reports
34// whichever it happened to read, which is a race wearing a registry's clothes.
35//
36// license_tier: ORIGINAL No hw writes (Rule 26). expect_exit: 0
37import "nx_syscalls.nx"
38import "nx_store_seed_lib.nx"
39
40const CWP_PREFIX: *u8 = "knowledge/store/cronwatch-"
41const CWP_ROOT: *u8 = "/volume1/homes/elderwesto/nishihost/"
42const CWP_ROWCAP: i64 = 4096
43// NOT A BUDGET -- A REFUSAL BOUNDARY. The plane's size is genuinely unknowable to a writer before it
44// reads it, so this is the one legitimate bound here. sts_load truncates ONLY when its buffer fills,
45// so a return STRICTLY LESS THAN cap PROVES completeness; n == cap is ambiguous and REFUSES loudly
46// rather than letting a duplicate slip past an unknowingly partial read.
47const CWP_PLANECAP: i64 = 8388608
48const CWP_TAB: i64 = 9
49const CWP_NL: i64 = 10
50const CWP_SLASH: i64 = 47
51const CWP_MINUS: i64 = 45
52const CWP_D0: i64 = 48
53const CWP_D9: i64 = 57
54const CWP_SPACE: i64 = 32
55const CWP_EXIT_USAGE: i64 = 2
56const CWP_EXIT_REFUSED: i64 = 1
57
58func cwp_len(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } return n }
59func cwp_out(s: *u8) -> i64 { sys_write(1, s, cwp_len(s)); return 0 }
60func cwp_err(s: *u8) -> i64 { sys_write(2, s, cwp_len(s)); return 0 }
61func cwp_cat(d: *u8, o: i64, s: *u8) -> i64 { var x: i64 = o; var i: i64 = 0; while s[i] != (0 as u8) { d[x] = s[i]; x = x + 1; i = i + 1 } return x }
62// A DIAGNOSTIC THAT CANNOT RENDER A NEGATIVE WILL HIDE ONE. Caught 2026-08-15 by this organ's own
63// first selftest run, which printed `got= want=` for the two sentinel cases: every tooth passed and
64// the evidence line was blank. The verdict vector was right and the thing a human reads was empty --
65// which is precisely the state in which a real failure would be unreadable.
66func cwp_num(d: *u8, o: i64, v: i64) -> i64 {
67 var x: i64 = o
68 var m: i64 = v
69 if m < 0 { d[x] = CWP_MINUS as u8; x = x + 1; m = 0 - m }
70 if m == 0 { d[x] = CWP_D0 as u8; return x + 1 }
71 let t: *u8 = sys_mmap(32)
72 var k: i64 = 0
73 while m > 0 { t[k] = (CWP_D0 + (m % 10)) as u8; m = m / 10; k = k + 1 }
74 var j: i64 = 0
75 while j < k { d[x] = t[k-1-j]; x = x + 1; j = j + 1 }
76 return x
77}
78func cwp_tabfree(s: *u8) -> i64 { var i: i64 = 0; while s[i] != (0 as u8) { if s[i] == (CWP_TAB as u8) { return 0 } i = i + 1 } return 1 }
79// strictly positive decimal, or -1. An empty string is NOT zero.
80func cwp_atoi(s: *u8) -> i64 {
81 if s[0] == (0 as u8) { return 0 - 1 }
82 var v: i64 = 0
83 var i: i64 = 0
84 while s[i] != (0 as u8) {
85 let c: i64 = s[i] as i64
86 if c < CWP_D0 { return 0 - 1 }
87 if c > CWP_D9 { return 0 - 1 }
88 v = v * 10 + (c - CWP_D0)
89 i = i + 1
90 }
91 return v
92}
93// relative paths resolve against the estate root exactly as the clock and the watcher resolve them
94func cwp_resolve(p: *u8, dst: *u8) -> i64 {
95 var o: i64 = 0
96 if p[0] != (CWP_SLASH as u8) { o = cwp_cat(dst, 0, CWP_ROOT) }
97 o = cwp_cat(dst, o, p)
98 dst[o] = 0 as u8
99 return o
100}
101func cwp_exists(p: *u8) -> i64 {
102 let path: *u8 = sys_mmap(CWP_ROWCAP)
103 cwp_resolve(p, path)
104 let sb: *u8 = sys_mmap(256)
105 if sys_fstatat(path, sb) < 0 { return 0 }
106 return 1
107}
108func cwp_contains(buf: *u8, n: i64, pat: *u8) -> i64 {
109 let pl: i64 = cwp_len(pat)
110 if pl <= 0 { return 0 }
111 var i: i64 = 0
112 var found: i64 = 0
113 var go: i64 = 1
114 while go == 1 {
115 if i + pl > n { go = 0 }
116 else {
117 var j: i64 = 0
118 var hit: i64 = 1
119 while j < pl { if buf[i+j] != pat[j] { hit = 0; j = pl } else { j = j + 1 } }
120 if hit == 1 { found = 1; go = 0 } else { i = i + 1 }
121 }
122 }
123 return found
124}
125// a row matches a name ONLY as the whole first column. A prefix match would let `growth` collide with
126// `growthbeat` and refuse a legitimate distinct watch, so the TAB is part of the test.
127func cwp_dup(plane: *u8, n: i64, name: *u8) -> i64 {
128 let nl: i64 = cwp_len(name)
129 if nl <= 0 { return 0 }
130 var s: i64 = 0
131 var found: i64 = 0
132 var go: i64 = 1
133 while go == 1 {
134 if s >= n { go = 0 }
135 else {
136 var e: i64 = s
137 var g2: i64 = 1
138 while g2 == 1 { if e >= n { g2 = 0 } else { if plane[e] == (CWP_NL as u8) { g2 = 0 } else { e = e + 1 } } }
139 if e - s > nl {
140 var j: i64 = 0
141 var hit: i64 = 1
142 while j < nl { if plane[s+j] != name[j] { hit = 0; j = nl } else { j = j + 1 } }
143 if hit == 1 { if plane[s+nl] == (CWP_TAB as u8) { found = 1; go = 0 } }
144 }
145 if go == 1 { s = e + 1 }
146 }
147 }
148 return found
149}
150
151func cwp_t(nm: *u8, got: i64, want: i64, ctr: *i64) -> i64 {
152 ctr[1] = ctr[1] + 1
153 cwp_out(" " as *u8)
154 if got == want { ctr[0] = ctr[0] + 1; cwp_out("PASS " as *u8) } else { cwp_out("FAIL " as *u8) }
155 cwp_out(nm)
156 let b: *u8 = sys_mmap(CWP_ROWCAP)
157 var o: i64 = cwp_cat(b, 0, " got=" as *u8)
158 o = cwp_num(b, o, got)
159 o = cwp_cat(b, o, " want=" as *u8)
160 o = cwp_num(b, o, want)
161 b[o] = CWP_NL as u8; o = o + 1
162 sys_write(1, b, o)
163 return 0
164}
165// fixtures assembled from bytes at runtime -- a scanner that reads source finds its own fixture.
166func cwp_plane_fixture(d: *u8) -> i64 {
167 var p: i64 = cwp_cat(d, 0, "growthbeat" as *u8)
168 d[p] = CWP_TAB as u8; p = p + 1
169 p = cwp_cat(d, p, "a" as *u8)
170 d[p] = CWP_NL as u8; p = p + 1
171 p = cwp_cat(d, p, "netobs-beat" as *u8)
172 d[p] = CWP_TAB as u8; p = p + 1
173 p = cwp_cat(d, p, "b" as *u8)
174 d[p] = CWP_NL as u8; p = p + 1
175 d[p] = 0 as u8
176 return p
177}
178func cwp_selftest() -> i64 {
179 let ctr: *i64 = sys_mmap(16) as *i64
180 ctr[0] = 0
181 ctr[1] = 0
182 cwp_out("nx_cronwatch_put selftest\n" as *u8)
183
184 let tabbed: *u8 = sys_mmap(64)
185 var tp: i64 = cwp_cat(tabbed, 0, "ab" as *u8)
186 tabbed[tp] = CWP_TAB as u8; tp = tp + 1
187 tp = cwp_cat(tabbed, tp, "cd" as *u8)
188 tabbed[tp] = 0 as u8
189 cwp_t("neg-control-tab-in-field-refused" as *u8, cwp_tabfree(tabbed), 0, ctr)
190 cwp_t("clean-field-accepted" as *u8, cwp_tabfree("abcd" as *u8), 1, ctr)
191
192 cwp_t("maxage-positive-parses" as *u8, cwp_atoi("27000" as *u8), 27000, ctr)
193 cwp_t("neg-control-maxage-nonnumeric-refused" as *u8, cwp_atoi("6h" as *u8), 0 - 1, ctr)
194 // AN EMPTY STRING IS NOT ZERO -- if it parsed as 0 the positivity check would be the only thing
195 // standing between an empty argv and a permanently-stale watch.
196 cwp_t("neg-control-maxage-empty-refused" as *u8, cwp_atoi("" as *u8), 0 - 1, ctr)
197
198 let pl: *u8 = sys_mmap(CWP_ROWCAP)
199 let pn: i64 = cwp_plane_fixture(pl)
200 cwp_t("dup-detects-existing-name" as *u8, cwp_dup(pl, pn, "growthbeat" as *u8), 1, ctr)
201 cwp_t("dup-allows-new-name" as *u8, cwp_dup(pl, pn, "compare-beat" as *u8), 0, ctr)
202 // THE COLUMN BOUNDARY IS PART OF THE TEST: a prefix match would refuse a legitimate distinct name.
203 cwp_t("neg-control-dup-is-not-a-prefix-match" as *u8, cwp_dup(pl, pn, "growth" as *u8), 0, ctr)
204 cwp_t("neg-control-dup-on-empty-plane" as *u8, cwp_dup(pl, 0, "growthbeat" as *u8), 0, ctr)
205
206 let hay: *u8 = "GROWTH epoch=1786817064 frontier_done=252 verdict=GREEN" as *u8
207 let hn: i64 = cwp_len(hay)
208 cwp_t("marker-found-when-present" as *u8, cwp_contains(hay, hn, "epoch=" as *u8), 1, ctr)
209 cwp_t("neg-control-marker-absent-refused" as *u8, cwp_contains(hay, hn, "ts=" as *u8), 0, ctr)
210
211 let rp: *u8 = sys_mmap(CWP_ROWCAP)
212 cwp_resolve("knowledge/status/x" as *u8, rp)
213 cwp_t("relative-path-resolves-under-root" as *u8, cwp_contains(rp, cwp_len(rp), "nishihost/knowledge/status/x" as *u8), 1, ctr)
214 cwp_resolve("/tmp/x" as *u8, rp)
215 cwp_t("absolute-path-passes-through" as *u8, cwp_contains(rp, cwp_len(rp), CWP_ROOT), 0, ctr)
216 // the guard must FIRE on a file that is not there, or it is not a guard
217 cwp_t("neg-control-missing-heartbeat-file" as *u8, cwp_exists("knowledge/status/__nx_cronwatch_no_such_file" as *u8), 0, ctr)
218
219 let b: *u8 = sys_mmap(CWP_ROWCAP)
220 var o: i64 = cwp_cat(b, 0, "passed " as *u8)
221 o = cwp_num(b, o, ctr[0])
222 b[o] = CWP_SLASH as u8; o = o + 1
223 o = cwp_num(b, o, ctr[1])
224 if ctr[0] == ctr[1] { o = cwp_cat(b, o, " verdict=GREEN" as *u8) } else { o = cwp_cat(b, o, " verdict=RED" as *u8) }
225 b[o] = CWP_NL as u8; o = o + 1
226 sys_write(1, b, o)
227 if ctr[0] == ctr[1] { return 0 }
228 return 1
229}
230
231func main(argc: i64, argv: *i64) -> i64 {
232 if argc < 2 { cwp_err("usage: nx_cronwatch_put put <name> <heartbeat-file> <ts-marker> <max-age-secs> <remediation> | nx_cronwatch_put selftest\n" as *u8); sys_exit(CWP_EXIT_USAGE); return CWP_EXIT_USAGE }
233 let verb: *u8 = argv[1] as *u8
234 if cwp_contains(verb, cwp_len(verb), "selftest" as *u8) == 1 { let r: i64 = cwp_selftest(); sys_exit(r); return r }
235 if cwp_contains(verb, cwp_len(verb), "put" as *u8) == 0 { cwp_err("nx_cronwatch_put: unknown verb -- only put and selftest\n" as *u8); sys_exit(CWP_EXIT_USAGE); return CWP_EXIT_USAGE }
236 if argc < 7 { cwp_err("nx_cronwatch_put: usage put <name> <heartbeat-file> <ts-marker> <max-age-secs> <remediation>\n" as *u8); sys_exit(CWP_EXIT_USAGE); return CWP_EXIT_USAGE }
237
238 let name: *u8 = argv[2] as *u8
239 let hb: *u8 = argv[3] as *u8
240 let mark: *u8 = argv[4] as *u8
241 let ages: *u8 = argv[5] as *u8
242 let rem: *u8 = argv[6] as *u8
243
244 if cwp_len(name) == 0 { cwp_err("nx_cronwatch_put: REFUSED empty name -- a nameless watch cannot be reported against\n" as *u8); sys_exit(CWP_EXIT_REFUSED); return CWP_EXIT_REFUSED }
245 if cwp_len(hb) == 0 { cwp_err("nx_cronwatch_put: REFUSED empty heartbeat-file\n" as *u8); sys_exit(CWP_EXIT_REFUSED); return CWP_EXIT_REFUSED }
246 if cwp_len(mark) == 0 { cwp_err("nx_cronwatch_put: REFUSED empty ts-marker\n" as *u8); sys_exit(CWP_EXIT_REFUSED); return CWP_EXIT_REFUSED }
247 if cwp_len(rem) == 0 { cwp_err("nx_cronwatch_put: REFUSED empty remediation -- a stale alarm with no remedy is a puzzle handed to whoever is on call\n" as *u8); sys_exit(CWP_EXIT_REFUSED); return CWP_EXIT_REFUSED }
248 if cwp_tabfree(name) == 0 { cwp_err("nx_cronwatch_put: REFUSED tab inside name -- it would split into phantom columns\n" as *u8); sys_exit(CWP_EXIT_REFUSED); return CWP_EXIT_REFUSED }
249 if cwp_tabfree(hb) == 0 { cwp_err("nx_cronwatch_put: REFUSED tab inside heartbeat-file\n" as *u8); sys_exit(CWP_EXIT_REFUSED); return CWP_EXIT_REFUSED }
250 if cwp_tabfree(mark) == 0 { cwp_err("nx_cronwatch_put: REFUSED tab inside ts-marker\n" as *u8); sys_exit(CWP_EXIT_REFUSED); return CWP_EXIT_REFUSED }
251 if cwp_tabfree(rem) == 0 { cwp_err("nx_cronwatch_put: REFUSED tab inside remediation\n" as *u8); sys_exit(CWP_EXIT_REFUSED); return CWP_EXIT_REFUSED }
252
253 let age: i64 = cwp_atoi(ages)
254 if age < 1 { cwp_err("nx_cronwatch_put: REFUSED max-age must be a positive integer of seconds (expected-interval + grace; interval + interval/4 absorbs jitter and still catches the first missed fire)\n" as *u8); sys_exit(CWP_EXIT_REFUSED); return CWP_EXIT_REFUSED }
255
256 if cwp_exists(hb) == 0 { cwp_err("nx_cronwatch_put: REFUSED heartbeat file does not exist -- a watch over a missing file reports STALE from the moment it is written and never stops, which trains the reader to ignore the whole board\n" as *u8); sys_exit(CWP_EXIT_REFUSED); return CWP_EXIT_REFUSED }
257
258 let hbp: *u8 = sys_mmap(CWP_ROWCAP)
259 cwp_resolve(hb, hbp)
260 let hlen: *i64 = sys_mmap(16) as *i64
261 let hbuf: *u8 = sys_read_file(hbp, hlen)
262 if (hbuf as i64) == 0 { cwp_err("nx_cronwatch_put: REFUSED heartbeat file exists but could not be read\n" as *u8); sys_exit(CWP_EXIT_REFUSED); return CWP_EXIT_REFUSED }
263 if cwp_contains(hbuf, hlen[0], mark) == 0 { cwp_err("nx_cronwatch_put: REFUSED ts-marker does not appear anywhere in the heartbeat file -- the watch would report STALE from birth. Check the marker against a real line of that log before declaring it.\n" as *u8); sys_exit(CWP_EXIT_REFUSED); return CWP_EXIT_REFUSED }
264
265 let plane: *u8 = sys_mmap(CWP_PLANECAP)
266 let pn: i64 = sts_load(CWP_PREFIX, plane, CWP_PLANECAP)
267 if pn >= CWP_PLANECAP { cwp_err("nx_cronwatch_put: REFUSED plane read filled its buffer, so completeness cannot be proven and a duplicate could slip past a partial read\n" as *u8); sys_exit(CWP_EXIT_REFUSED); return CWP_EXIT_REFUSED }
268 if pn > 0 { if cwp_dup(plane, pn, name) == 1 { cwp_err("nx_cronwatch_put: REFUSED a watch of that name already exists -- one subject per row, or the check reports whichever row it happened to read\n" as *u8); sys_exit(CWP_EXIT_REFUSED); return CWP_EXIT_REFUSED } }
269
270 let row: *u8 = sys_mmap(CWP_ROWCAP)
271 var o: i64 = 0
272 o = cwp_cat(row, o, name); row[o] = CWP_TAB as u8; o = o + 1
273 o = cwp_cat(row, o, hb); row[o] = CWP_TAB as u8; o = o + 1
274 o = cwp_cat(row, o, mark); row[o] = CWP_TAB as u8; o = o + 1
275 o = cwp_num(row, o, age); row[o] = CWP_TAB as u8; o = o + 1
276 o = cwp_cat(row, o, rem)
277 if o >= CWP_ROWCAP - 2 { cwp_err("nx_cronwatch_put: REFUSED row exceeds CWP_ROWCAP rather than truncating it into a malformed row\n" as *u8); sys_exit(CWP_EXIT_REFUSED); return CWP_EXIT_REFUSED }
278
279 let rc: i64 = sts_append_fast_locked(CWP_PREFIX, row, o)
280 if rc < 0 { cwp_err("nx_cronwatch_put: FAIL commit error\n" as *u8); sys_exit(CWP_EXIT_REFUSED); return CWP_EXIT_REFUSED }
281
282 let msg: *u8 = sys_mmap(CWP_ROWCAP)
283 var mo: i64 = cwp_cat(msg, 0, "CRONWATCH-PUT ok rows=" as *u8)
284 mo = cwp_num(msg, mo, rc)
285 mo = cwp_cat(msg, mo, " plane_bytes_before=" as *u8)
286 mo = cwp_num(msg, mo, pn)
287 mo = cwp_cat(msg, mo, " row=" as *u8)
288 mo = cwp_cat(msg, mo, row)
289 mo = cwp_cat(msg, mo, " -- VERIFY WITH THE CONSUMER, NEVER THIS RECEIPT: nx_cron_watch check\n" as *u8)
290 sys_write(1, msg, mo)
291 sys_exit(0)
292 return 0
293}