nx_heartbeat_monitor.nx source
↩ module page · 201 lines · 9588 B
1// nx_heartbeat_monitor.nx -- WMS-M1: the LIVENESS / HEARTBEAT MONITOR.
2//
3// module: nishi-core.wms.heartbeat_monitor
4// capability: CORE_COMPUTE (durable cross-stream liveness detector built ON WMS-R0)
5//
6// WHAT THIS CLOSES: the autonomous conductor/team-pulse loop can CRASH mid-run and
7// nobody notices -- the work just silently stops. There was no durable detector that
8// could span streams AND survive a crash: nx_progress_watchdog.nx is in-RAM,
9// single-process, monotonic (its STALLED/OK state dies with the process and cannot
10// see a sibling stream). nx_heartbeat_register.nx is capability-registry wiring, not
11// a monitor. M1 reuses ONLY the STALLED/OK semantics of the watchdog; the durable,
12// cross-stream substrate is the WMS-R0 atomic framed-append log (a heartbeat channel).
13//
14// THE MECHANISM: every active workstream BEATS an epoch -- one framed record appended
15// (via the WMS-R0 locked single-write primitive) into a shared heartbeat channel:
16// "HBX ws=<id> epoch=<beatsec> seq=<n> actor=<pid> END"
17// A monitor REPLAYS the channel and, for a set of ws-ids, computes age = now - last_beat
18// and flags STALLED iff (the stream has beaten AND age > threshold). Last-writer-by-
19// MAX-epoch makes it robust to out-of-order append: a fresh beat ALWAYS clears a stale
20// reading (this is what makes the TAMPER/resume lane clear correctly).
21//
22// REUSE / lineage (check-registries-before-building):
23// - fa_append / fa_appendz / fa_cat / fa_catn (nx_framed_append.nx, WMS-R0b): the
24// atomic locked single-write framing primitive -> each beat = exactly one write().
25// - wsl_at / wsl_field / the per-'\n' line-scan replay skeleton (nx_ws_ledger.nx,
26// WMS-R2): token match + decimal field extract + no-silent-skip flagging.
27// Sovereign: only nx_syscalls + nx_framed_append + nx_ws_ledger. Additive: no organ
28// is modified -- new files only.
29//
30// WRITE DISCIPLINE (the torn-line bug we just killed): EVERY heartbeat record is
31// assembled into ONE buffer (fa_cat/fa_catn) and emitted with a SINGLE fa_append
32// locked write. NEVER a sequence of sys_write() calls.
33// license_tier: ORIGINAL
34import "nx_syscalls.nx"
35import "nx_framed_append.nx"
36import "nx_ws_ledger.nx"
37const HB_MAGIC_2277: i64 = 2277
38
39const HB_RECCAP: i64 = 256 // bounded record size (matches R0/R2 RECCAP)
40const HB_ALIVE: i64 = 0 // beating within threshold
41const HB_STALLED: i64 = 1 // last beat older than threshold (the crash signal)
42const HB_UNKNOWN: i64 = 2 // never seen a beat for this stream
43
44// Assemble ONE heartbeat record into buf (NO trailing newline); returns rec_len.
45// Mirrors wsl_build: leading "HBX " sentinel + trailing " END" anchor.
46// "HBX ws=<id> epoch=<beatsec> seq=<n> actor=<pid> END"
47func hbm_build(buf: *u8, ws: i64, epoch: i64, seq: i64, actor: i64) -> i64 {
48 var o: i64 = 0
49 o = fa_cat(buf, o, "HBX ws=\x00" as *u8)
50 o = fa_catn(buf, o, ws)
51 o = fa_cat(buf, o, " epoch=\x00" as *u8)
52 o = fa_catn(buf, o, epoch)
53 o = fa_cat(buf, o, " seq=\x00" as *u8)
54 o = fa_catn(buf, o, seq)
55 o = fa_cat(buf, o, " actor=\x00" as *u8)
56 o = fa_catn(buf, o, actor)
57 o = fa_cat(buf, o, " END\x00" as *u8)
58 return o
59}
60
61// TIME-INJECTABLE beat: caller supplies `epoch` (lets the gate place beats at
62// controlled times -- no hidden wall-clock). Builds into ONE buffer, then ONE
63// locked write via the R0 primitive. Returns fa_append's rc:
64// > 0 bytes written -1 open fail -2 oversized(rejected) -3 short write.
65func hb_beat_at(path: *u8, ws: i64, epoch: i64, seq: i64, actor: i64) -> i64 {
66 let buf: *u8 = sys_mmap(HB_RECCAP + 16)
67 let rl: i64 = hbm_build(buf, ws, epoch, seq, actor)
68 return fa_append(path, buf, rl, HB_RECCAP)
69}
70
71// PRODUCTION beat: a live workstream calls this each loop iteration. Stamps the
72// wall-clock epoch and the caller's pid as actor, then the single locked write.
73func hb_beat(path: *u8, ws: i64, seq: i64) -> i64 {
74 let epoch: i64 = sys_now_realtime_sec()
75 let pid: i64 = __syscall(172, 0, 0, 0, 0, 0, 0) // getpid (actor). rv64 getpid=172; raw x86 39 is an RV64 KEY translated to ioctl(16) -> -ENOTTY (debt idx HB_MAGIC_2277)
76 return hb_beat_at(path, ws, epoch, seq, pid)
77}
78
79// Well-formedness of heartbeat line [ls,le) (le = index of the '\n'): the 2-anchor
80// discipline -- begins "HBX " and the 4 bytes before '\n' are " END". A torn/corrupt
81// line fails this and is FLAGGED by the scan (never silently skipped). Returns 1/0.
82func hb_wellformed(b: *u8, ls: i64, le: i64) -> i64 {
83 let llen: i64 = le - ls
84 if llen < 8 { return 0 }
85 if b[ls] != (72 as u8) { return 0 } // H
86 if b[ls + 1] != (66 as u8) { return 0 } // B
87 if b[ls + 2] != (88 as u8) { return 0 } // X
88 if b[ls + 3] != (32 as u8) { return 0 } // space
89 let e: i64 = le - 1
90 if b[e - 3] != (32 as u8) { return 0 } // space
91 if b[e - 2] != (69 as u8) { return 0 } // E
92 if b[e - 1] != (78 as u8) { return 0 } // N
93 if b[e] != (68 as u8) { return 0 } // D
94 return 1
95}
96
97// LAST BEAT of stream `ws`: replay the whole channel, return the MAXIMUM epoch= of
98// any well-formed HBX record whose ws= matches. Last-writer-by-max-epoch -> robust
99// to out-of-order append: a later/fresher beat ALWAYS wins. Returns -1 if the
100// stream has never beaten (never-seen). Reuses wsl_field (token match + int parse).
101func hb_last_beat(path: *u8, ws: i64) -> i64 {
102 let szp: *i64 = sys_mmap(16) as *i64
103 let b: *u8 = sys_read_file(path, szp)
104 let sz: i64 = szp[0]
105 var best: i64 = 0 - 1
106 var ls: i64 = 0
107 var i: i64 = 0
108 while i < sz {
109 if b[i] == (10 as u8) {
110 if hb_wellformed(b, ls, i) == 1 {
111 let rws: i64 = wsl_field(b, ls, i, "ws=\x00" as *u8, 3)
112 if rws == ws {
113 let ep: i64 = wsl_field(b, ls, i, "epoch=\x00" as *u8, 6)
114 if ep >= 0 { if ep > best { best = ep } }
115 }
116 }
117 ls = i + 1
118 }
119 i = i + 1
120 }
121 return best
122}
123
124// THE DETECTOR. For each ws-id in ids[0..nids): last = hb_last_beat; age = now-last;
125// verdicts[k] = HB_STALLED iff (last >= 0 AND age > threshold)
126// HB_UNKNOWN iff (last < 0) (never beaten)
127// HB_ALIVE otherwise
128// Also scans the channel ONCE to FLAG (count) any malformed line -- corruption is
129// surfaced, never swallowed (ws_ledger no-silent-skip discipline).
130// outs[0] = stalled_count outs[1] = alive_count outs[2] = flagged_malformed
131// `now` and `threshold` are CALLER PARAMETERS (the gate controls them) -- no hidden
132// wall-clock, no magic number baked into the detector (Cardinal 11). Returns stalled.
133func hbm_scan(path: *u8, now: i64, threshold: i64, ids: *i64, nids: i64,
134 verdicts: *i64, outs: *i64) -> i64 {
135 var stalled: i64 = 0
136 var alive: i64 = 0
137 var k: i64 = 0
138 while k < nids {
139 let id: i64 = ids[k]
140 let last: i64 = hb_last_beat(path, id)
141 if last < 0 {
142 verdicts[k] = HB_UNKNOWN
143 } else {
144 let age: i64 = now - last
145 if age > threshold {
146 verdicts[k] = HB_STALLED
147 stalled = stalled + 1
148 } else {
149 verdicts[k] = HB_ALIVE
150 alive = alive + 1
151 }
152 }
153 k = k + 1
154 }
155 // single corruption sweep over the channel (flag, never drop)
156 let szp: *i64 = sys_mmap(16) as *i64
157 let b: *u8 = sys_read_file(path, szp)
158 let sz: i64 = szp[0]
159 var flagged: i64 = 0
160 var ls: i64 = 0
161 var i: i64 = 0
162 while i < sz {
163 if b[i] == (10 as u8) {
164 if hb_wellformed(b, ls, i) == 0 { flagged = flagged + 1 }
165 ls = i + 1
166 }
167 i = i + 1
168 }
169 outs[0] = stalled
170 outs[1] = alive
171 outs[2] = flagged
172 return stalled
173}
174
175// Self-test main: the harness RUNS the compiled organ, so prove the capability is
176// live end-to-end on its own (3 beats, one stale, scan flags exactly it). Primary
177// referee is the gate; this is a sovereign smoke-test so the organ never builds dead.
178func _hbm_p(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} sys_write(1,s,n); return 0 }
179func _hbm_n(v: i64) -> i64 { let bb: *u8=sys_mmap(28); var m: i64=v; if m<0{m=0-m; sys_write(1,"-\x00" as *u8,1)}; let t: *u8=sys_mmap(28); var k: i64=0; if m==0{t[0]=48 as u8;k=1}; while m>0{t[k]=(48+(m%10)) as u8;m=m/10;k=k+1}; var i: i64=0; while i<k{bb[i]=t[k-1-i];i=i+1}; sys_write(1,bb,k); return 0 }
180func main() -> i64 {
181 _hbm_p("=== nx_heartbeat_monitor self-test ===\n" as *u8)
182 let p: *u8 = "/tmp/hbm_selftest.log\x00" as *u8
183 let fr: i64 = sys_openat_wr(p, 0x1a4) // fresh channel
184 if fr > 0 { sys_close(fr) }
185 let pid: i64 = __syscall(172, 0, 0, 0, 0, 0, 0) // rv64 getpid=172 (was raw x86 39 -> ioctl -> -ENOTTY, debt idx HB_MAGIC_2277)
186 hb_beat_at(p, 1, 990, 0, pid) // fresh
187 hb_beat_at(p, 2, 985, 0, pid) // fresh
188 hb_beat_at(p, 3, 900, 0, pid) // stale at now=1000,thr=60
189 let ids: *i64 = sys_mmap(8 * 4) as *i64
190 ids[0] = 1; ids[1] = 2; ids[2] = 3
191 let vd: *i64 = sys_mmap(8 * 4) as *i64
192 let outs: *i64 = sys_mmap(32) as *i64
193 let st: i64 = hbm_scan(p, 1000, 60, ids, 3, vd, outs)
194 _hbm_p(" v1=\x00" as *u8); _hbm_n(vd[0])
195 _hbm_p(" v2=\x00" as *u8); _hbm_n(vd[1])
196 _hbm_p(" v3=\x00" as *u8); _hbm_n(vd[2])
197 _hbm_p(" stalled=\x00" as *u8); _hbm_n(st)
198 _hbm_p("\n\x00" as *u8)
199 sys_exit(0)
200 return 0
201}