code wiki / _hdl_build / nx_heartbeat_str.nx
nx_heartbeat_str.nx source
↩ module page · 203 lines · 8159 B
1// nx_heartbeat_str.nx -- WMS LIVE-2: STRING-KEYED heartbeats + registry-sourced monitoring.
2//
3// module: nishi-core.wms.heartbeat_str
4// capability: CORE_COMPUTE (auto-monitor every registered workstream from the SSOT)
5//
6// WHAT THIS CLOSES: M1 (nx_heartbeat_monitor) keys heartbeats by INTEGER ws, but the R1 registry
7// keys workstreams by STRING id ("T-WS-A") -- the int/string seam that blocked LIVE-2 (and showed up
8// at R9). Operator decision 2026-06-14: STRING-KEYED heartbeats, eliminating the seam. A stream now
9// beats under its REGISTRY id, and hbm_scan_registry sources the monitored set straight from ws:ids,
10// so EVERY registered empire is auto-covered with NO bridge/fabricated mapping.
11//
12// ADDITIVE (rule 19): the integer hb_* in nx_heartbeat_monitor are UNTOUCHED -- existing callers
13// (the LIVE-1 conductor, the M1 gate) keep working. This is a new, parallel string-keyed capability.
14//
15// REUSE (rule 15): the locked single-write framing (fa_append/fa_cat/fa_catn, WMS-R0b), the HBX
16// 2-anchor wellformed discipline (M1), and registry enumeration (ws_manifest_p, R1). The only new
17// machinery is a STRING field extractor (token up to the next space) + string compare -- M1's
18// wsl_field parses ints, which can't carry a string id. Sovereign: nx_syscalls + nx_framed_append +
19// nx_workstream_store. license_tier: ORIGINAL
20import "nx_syscalls.nx"
21import "nx_framed_append.nx"
22import "nx_workstream_store.nx"
23
24const HBS_RECCAP: i64 = 256
25const HBS_ALIVE: i64 = 0 // beating within threshold
26const HBS_STALLED: i64 = 1 // last beat older than threshold (the crash signal)
27const HBS_UNKNOWN: i64 = 2 // registered but never beaten
28
29func hbs_streq(a: *u8, b: *u8) -> i64 {
30 var i: i64 = 0
31 while a[i] != (0 as u8) { if a[i] != b[i] { return 0 } i = i + 1 }
32 if b[i] != (0 as u8) { return 0 }
33 return 1
34}
35
36// token match: does `pat` (plen) occur at b[pos..pos+plen) within [.,le)?
37func hbs_at(b: *u8, pos: i64, le: i64, pat: *u8, plen: i64) -> i64 {
38 if pos + plen > le { return 0 }
39 var k: i64 = 0
40 while k < plen { if b[pos + k] != pat[k] { return 0 } k = k + 1 }
41 return 1
42}
43
44// HBX line wellformed (le = index of '\n'): begins "HBX " AND the 4 bytes before '\n' are " END".
45func hbs_wellformed(b: *u8, ls: i64, le: i64) -> i64 {
46 let llen: i64 = le - ls
47 if llen < 8 { return 0 }
48 if b[ls] != (72 as u8) { return 0 } // H
49 if b[ls + 1] != (66 as u8) { return 0 } // B
50 if b[ls + 2] != (88 as u8) { return 0 } // X
51 if b[ls + 3] != (32 as u8) { return 0 } // space
52 let e: i64 = le - 1
53 if b[e - 3] != (32 as u8) { return 0 } // space
54 if b[e - 2] != (69 as u8) { return 0 } // E
55 if b[e - 1] != (78 as u8) { return 0 } // N
56 if b[e] != (68 as u8) { return 0 } // D
57 return 1
58}
59
60// extract the STRING value of `key` within [ls,le): bytes after the key token up to the next space,
61// copied NUL-term into out. Returns length, or -1 if key absent.
62func hbs_field_str(b: *u8, ls: i64, le: i64, key: *u8, klen: i64, out: *u8) -> i64 {
63 var p: i64 = ls
64 while p < le {
65 if hbs_at(b, p, le, key, klen) == 1 {
66 var q: i64 = p + klen
67 var o: i64 = 0
68 while q < le {
69 if b[q] == (32 as u8) { q = le } else { out[o] = b[q]; o = o + 1; q = q + 1 }
70 }
71 out[o] = 0 as u8
72 return o
73 }
74 p = p + 1
75 }
76 out[0] = 0 as u8
77 return 0 - 1
78}
79
80// extract the INTEGER value of `key` within [ls,le): digits after the key token. -1 if absent.
81func hbs_field_int(b: *u8, ls: i64, le: i64, key: *u8, klen: i64) -> i64 {
82 var p: i64 = ls
83 while p < le {
84 if hbs_at(b, p, le, key, klen) == 1 {
85 var v: i64 = 0
86 var q: i64 = p + klen
87 while q < le {
88 let c: i64 = b[q] as i64
89 if c < 48 { q = le }
90 if c >= 48 { if c > 57 { q = le } if c <= 57 { v = v * 10 + (c - 48); q = q + 1 } }
91 }
92 return v
93 }
94 p = p + 1
95 }
96 return 0 - 1
97}
98
99// TIME-INJECTABLE string-keyed beat (the gate controls epoch). ONE locked framed write.
100// "HBX ws=<ws_str> epoch=<e> seq=<n> actor=<pid> END" (ws_str must be space-free, like a reg id)
101func hb_beat_s_at(path: *u8, ws_str: *u8, epoch: i64, seq: i64, actor: i64) -> i64 {
102 let buf: *u8 = sys_mmap(HBS_RECCAP + 16)
103 var o: i64 = 0
104 o = fa_cat(buf, o, "HBX ws=" as *u8); o = fa_cat(buf, o, ws_str)
105 o = fa_cat(buf, o, " epoch=" as *u8); o = fa_catn(buf, o, epoch)
106 o = fa_cat(buf, o, " seq=" as *u8); o = fa_catn(buf, o, seq)
107 o = fa_cat(buf, o, " actor=" as *u8); o = fa_catn(buf, o, actor)
108 o = fa_cat(buf, o, " END" as *u8)
109 buf[o] = 0 as u8
110 return fa_appendz(path, buf, HBS_RECCAP)
111}
112
113// PRODUCTION string-keyed beat: a live workstream beats under its REGISTRY id each loop iteration.
114func hb_beat_s(path: *u8, ws_str: *u8, seq: i64) -> i64 {
115 let epoch: i64 = sys_now_realtime_sec()
116 let pid: i64 = __syscall(39, 0, 0, 0, 0, 0, 0)
117 return hb_beat_s_at(path, ws_str, epoch, seq, pid)
118}
119
120// MAX epoch of any wellformed HBX record whose ws= token equals ws_str; -1 if never beaten.
121func hb_last_beat_s(path: *u8, ws_str: *u8) -> i64 {
122 let szp: *i64 = sys_mmap(16) as *i64
123 let b: *u8 = sys_read_file(path, szp)
124 let sz: i64 = szp[0]
125 let tok: *u8 = sys_mmap(256)
126 var best: i64 = 0 - 1
127 var ls: i64 = 0
128 var i: i64 = 0
129 while i < sz {
130 if b[i] == (10 as u8) {
131 if hbs_wellformed(b, ls, i) == 1 {
132 if hbs_field_str(b, ls, i, "ws=" as *u8, 3, tok) >= 0 {
133 if hbs_streq(tok, ws_str) == 1 {
134 let ep: i64 = hbs_field_int(b, ls, i, "epoch=" as *u8, 6)
135 if ep >= 0 { if ep > best { best = ep } }
136 }
137 }
138 }
139 ls = i + 1
140 }
141 i = i + 1
142 }
143 return best
144}
145
146// THE LIVE-2 CAPABILITY: source the monitored set from the R1 registry (ws:ids under `prefix`) and,
147// for EVERY registered workstream, compute liveness from the string-keyed heartbeat channel.
148// verdicts[k] = HBS_ALIVE / HBS_STALLED / HBS_UNKNOWN for the k-th registered id; out_ids[k] = its
149// NUL-term id ptr (so callers correlate by id, not array position). Also flags malformed lines.
150// outs[0]=stalled outs[1]=alive outs[2]=flagged outs[3]=unknown outs[4]=nids. Returns nids (the
151// coverage count = every registered empire, no hardcoded id list, no int/string bridge).
152func hbm_scan_registry(hbpath: *u8, prefix: *u8, now: i64, threshold: i64,
153 out_ids: *i64, verdicts: *i64, outs: *i64, cap: i64) -> i64 {
154 let ids: *i64 = sys_mmap(8 * cap) as *i64
155 let nids: i64 = ws_manifest_p(prefix, ids, cap)
156 var stalled: i64 = 0
157 var alive: i64 = 0
158 var unknown: i64 = 0
159 if nids > 0 {
160 var k: i64 = 0
161 while k < nids {
162 let id: *u8 = ids[k] as *u8
163 out_ids[k] = id as i64
164 let last: i64 = hb_last_beat_s(hbpath, id)
165 if last < 0 {
166 verdicts[k] = HBS_UNKNOWN
167 unknown = unknown + 1
168 } else {
169 let age: i64 = now - last
170 if age > threshold {
171 verdicts[k] = HBS_STALLED
172 stalled = stalled + 1
173 } else {
174 verdicts[k] = HBS_ALIVE
175 alive = alive + 1
176 }
177 }
178 k = k + 1
179 }
180 }
181 // corruption sweep over the channel (flag, never drop)
182 let szp: *i64 = sys_mmap(16) as *i64
183 let b: *u8 = sys_read_file(hbpath, szp)
184 let sz: i64 = szp[0]
185 var flagged: i64 = 0
186 var ls: i64 = 0
187 var i: i64 = 0
188 while i < sz {
189 if b[i] == (10 as u8) {
190 if hbs_wellformed(b, ls, i) == 0 { flagged = flagged + 1 }
191 ls = i + 1
192 }
193 i = i + 1
194 }
195 outs[0] = stalled
196 outs[1] = alive
197 outs[2] = flagged
198 outs[3] = unknown
199 var nout: i64 = nids
200 if nout < 0 { nout = 0 }
201 outs[4] = nout
202 return nout
203}