code wiki / _hdl_build / nx_conductor_live_wms_gate.nx
`nx_conductor_live_wms_gate.nx` source

↩ module page · 238 lines · 9806 B
1// nx_conductor_live_wms_gate.nx -- the REFEREE for WMS LIVE-1 + LIVE-3 (conductor adoption).
2//
3// LIVE-1 (conductor heartbeat): the live conductor now beats "HBX ws=1 ... END" each loop iteration
4//   via a single locked fa_appendz (cl_hb_beat). This lane proves that heartbeat is M1-detectable:
5//   a freshly-beaten conductor ws reads HB_ALIVE, and a conductor that stopped beating (aged beat)
6//   reads HB_STALLED -- so a REAL crash surfaces, not just a test channel. (NEG side: STALLED proves
7//   the detector isn't a constant-ALIVE.)
8// LIVE-3 (event-log torn-write kill): cl_event was 8 sequential sys_write calls to one O_APPEND fd;
9//   it is now ONE assembled locked fa_appendz. GOOD lane = 16 workers emit the migrated cl_event shape
10//   concurrently -> ZERO torn lines, all present. BAD lane (mandatory NEG-CONTROL) = the SAME records
11//   emitted the OLD multi-write way -> concurrent interleave -> torn lines MUST appear. TAMPER: an
12//   oversized record is REJECTED (-2).
13//
14// A cl_event line is WELL-FORMED iff it starts "beat=" AND contains exactly one "beat=" head AND
15// exactly one " ms=" field token -- any interleaving duplicates/breaks those. GREEN(0) iff: good_torn==0
16// AND good_lines==N*NRECS AND bad_torn>0 AND tamper==-2 AND conductor ALIVE AND crashed STALLED.
17// Sovereign: nx_heartbeat_monitor + nx_framed_append + nx_syscalls.
18// license_tier: ORIGINAL
19import "nx_heartbeat_monitor.nx"
20import "nx_framed_append.nx"
21import "nx_syscalls.nx"
22
23const NWORKERS: i64 = 16
24const NRECS:    i64 = 200
25const RECCAP:   i64 = 256
26const CLG_LOG: *u8 = "knowledge/status/conductor_live_wms_gate.log"
27const GOODP: *u8 = "/tmp/clg_good.log\x00" as *u8
28const BADP:  *u8 = "/tmp/clg_bad.log\x00" as *u8
29
30// single-sink writers: write ONLY to the passed fd (stdout via fd 1). The durable evidence is the
31// one assembled fa_appendz line at the end -- so these never double-write.
32func gp(fd: i64, s: *u8) -> i64 {
  var n: i64 = 0
  while s[n] != (0 as u8) { n = n + 1 }
  sys_write(fd, s, n)
  return 0
37}
38func gn(fd: i64, v: i64) -> i64 {
  let bb: *u8 = sys_mmap(28)
  var m: i64 = v
  if m < 0 { sys_write(fd, "-\x00" as *u8, 1); m = 0 - m }
  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(fd, bb, k)
  return 0
50}
51
52// GOOD worker: the MIGRATED cl_event shape via a single locked fa_appendz.
53func good_worker(wid: i64) -> i64 {
  let pid: i64 = __syscall(39, 0, 0, 0, 0, 0, 0)
  let buf: *u8 = sys_mmap(RECCAP + 16)
  var i: i64 = 0
  while i < NRECS {
      var o: i64 = 0
      o = fa_cat(buf, o, "beat=\x00" as *u8); o = fa_catn(buf, o, i)
      o = fa_cat(buf, o, " tick=author : superopt PROPOSE/SCORE/VERIFY/EMIT(mem)  \x00" as *u8)
      o = fa_cat(buf, o, " verdict=PROVEN \x00" as *u8)
      o = fa_cat(buf, o, " ms=\x00" as *u8); o = fa_catn(buf, o, wid)
      fa_appendz(GOODP, buf, RECCAP)
      i = i + 1
  }
  return 0
67}
68
69func bad_num(fd: i64, v: i64) -> i64 {
  let nb: *u8 = sys_mmap(28)
  var m: i64 = v
  var l: i64 = 0
  if m == 0 { nb[0] = 48 as u8; l = 1 }
  while m > 0 { nb[l] = (48 + (m % 10)) as u8; m = m / 10; l = l + 1 }
  let d: *u8 = sys_mmap(28)
  var x: i64 = 0
  while x < l { d[x] = nb[l - 1 - x]; x = x + 1 }
  sys_write(fd, d, l)
  return 0
80}
81
82// BAD worker (NEG-CONTROL): the ORIGINAL cl_event -- a SEQUENCE of separate sys_write calls.
83func bad_worker(wid: i64) -> i64 {
  let fd: i64 = sys_openat_append(BADP, 0x1a4)
  if fd < 0 { return 0 - 1 }
  var i: i64 = 0
  while i < NRECS {
      sys_write(fd, "beat=\x00" as *u8, 5)
      bad_num(fd, i)
      sys_write(fd, " tick=author : superopt PROPOSE/SCORE/VERIFY/EMIT(mem)  \x00" as *u8, 56)
      sys_write(fd, " verdict=PROVEN \x00" as *u8, 15)
      sys_write(fd, " ms=\x00" as *u8, 4)
      bad_num(fd, wid)
      sys_write(fd, "\n\x00" as *u8, 1)
      i = i + 1
  }
  sys_close(fd)
  return 0
99}
100
101func spawn_all(which: i64) -> i64 {
  let pids: *i64 = sys_mmap(8 * (NWORKERS + 4)) as *i64
  var w: i64 = 0
  while w < NWORKERS {
      let pid: i64 = sys_fork()
      if pid == 0 {
          if which == 0 { good_worker(w) }
          if which == 1 { bad_worker(w) }
          sys_exit(0)
      }
      pids[w] = pid
      w = w + 1
  }
  let st: *i64 = sys_mmap(16) as *i64
  w = 0
  while w < NWORKERS { sys_wait4(pids[w], st, 0); w = w + 1 }
  return 0
118}
119
120// substring count of `pat` (pl bytes) within [a,b) of hay
121func slice_count(hay: *u8, a: i64, b: i64, pat: *u8, pl: i64) -> i64 {
  var c: i64 = 0
  var i: i64 = a
  while i + pl <= b {
      var k: i64 = 0
      var hit: i64 = 1
      while k < pl { if hay[i + k] != pat[k] { hit = 0; k = pl } else { k = k + 1 } }
      if hit == 1 { c = c + 1 }
      i = i + 1
  }
  return c
132}
133
134// total_lines into outs[0]; returns torn count. WELL-FORMED iff starts "beat=" AND exactly one
135// "beat=" head AND exactly one " ms=" field token in the line.
136func count_torn(path: *u8, outs: *i64) -> i64 {
  let szp: *i64 = sys_mmap(16) as *i64
  let b: *u8 = sys_read_file(path, szp)
  let sz: i64 = szp[0]
  var torn: i64 = 0
  var lines: i64 = 0
  var ls: i64 = 0
  var i: i64 = 0
  let head: *u8 = "beat=\x00" as *u8     // 5 bytes
  let msf:  *u8 = " ms=\x00" as *u8      // 4 bytes
  while i < sz {
      if b[i] == (10 as u8) {
          let llen: i64 = i - ls
          lines = lines + 1
          var ok: i64 = 1
          if llen < 10 { ok = 0 }
          if ok == 1 {
              var p: i64 = 0
              while p < 5 { if b[ls + p] != head[p] { ok = 0 } p = p + 1 }
          }
          if ok == 1 { if slice_count(b, ls, i, head, 5) != 1 { ok = 0 } }
          if ok == 1 { if slice_count(b, ls, i, msf, 4) != 1 { ok = 0 } }
          if ok == 0 { torn = torn + 1 }
          ls = i + 1
      }
      i = i + 1
  }
  outs[0] = lines
  return torn
165}
166
167func main() -> i64 {
  let logfd: i64 = sys_openat_append(CLG_LOG, 0x1a4)
169
  // fresh files
  let g0: i64 = sys_openat_wr(GOODP, 0x1a4); if g0 > 0 { sys_close(g0) }
  let b0: i64 = sys_openat_wr(BADP, 0x1a4); if b0 > 0 { sys_close(b0) }
173
  // ===== LIVE-3: concurrent migrated cl_event vs old multi-write =====
  spawn_all(0)
  spawn_all(1)
  let outs: *i64 = sys_mmap(16) as *i64
  let good_torn: i64 = count_torn(GOODP, outs)
  let good_lines: i64 = outs[0]
  let bad_torn: i64 = count_torn(BADP, outs)
  let bad_lines: i64 = outs[0]
  let want_lines: i64 = NWORKERS * NRECS
183
  // TAMPER: oversized record rejected with -2
  let trec: *u8 = sys_mmap(64)
  var ti: i64 = 0
  while ti < 32 { trec[ti] = 65 as u8; ti = ti + 1 }
  let tamper_rc: i64 = fa_append("/tmp/clg_tamper.log\x00" as *u8, trec, 32, 8)
189
  // ===== LIVE-1: conductor heartbeat is M1-detectable (ALIVE fresh / STALLED on crash) =====
  let hbp: *u8 = "/tmp/clg_hb.log\x00" as *u8
  let h0: i64 = sys_openat_wr(hbp, 0x1a4); if h0 > 0 { sys_close(h0) }
  let pid: i64 = __syscall(39, 0, 0, 0, 0, 0, 0)
  hb_beat_at(hbp, 1, 990, 0, pid)        // conductor ws=1 beating recently
  hb_beat_at(hbp, 9, 900, 0, pid)        // a CRASHED conductor (ws=9) -- last beat long ago
  let ids: *i64 = sys_mmap(8 * 4) as *i64
  ids[0] = 1; ids[1] = 9
  let vd: *i64 = sys_mmap(8 * 4) as *i64
  let houts: *i64 = sys_mmap(32) as *i64
  hbm_scan(hbp, 1000, 60, ids, 2, vd, houts)   // now=1000, threshold=60
  var live_ok: i64 = 0
  if vd[0] == HB_ALIVE { if vd[1] == HB_STALLED { live_ok = 1 } }
203
  var green: i64 = 1
  if good_lines != want_lines { green = 0 }
  if good_torn != 0 { green = 0 }
  if bad_torn <= 0 { green = 0 }
  if tamper_rc != (0 - 2) { green = 0 }
  if live_ok != 1 { green = 0 }
210
  gp(1, "WMS LIVE-1+LIVE-3 conductor-adoption gate\n" as *u8)
  gp(1, "  LIVE-3 good_lines=\x00" as *u8); gn(1, good_lines); gp(1, "/\x00" as *u8); gn(1, want_lines)
  gp(1, " good_torn=\x00" as *u8); gn(1, good_torn)
  if good_torn == 0 { if good_lines == want_lines { gp(1, " PASS\n\x00" as *u8) } else { gp(1, " FAIL\n\x00" as *u8) } } else { gp(1, " FAIL\n\x00" as *u8) }
  gp(1, "  LIVE-3 neg-control bad_torn=\x00" as *u8); gn(1, bad_torn)
  if bad_torn > 0 { gp(1, " PASS(detector-sees-tearing)\n\x00" as *u8) } else { gp(1, " FAIL(no-tearing)\n\x00" as *u8) }
  gp(1, "  LIVE-3 tamper=\x00" as *u8); gn(1, tamper_rc)
  if tamper_rc == (0 - 2) { gp(1, " PASS\n\x00" as *u8) } else { gp(1, " FAIL\n\x00" as *u8) }
  gp(1, "  LIVE-1 conductor_ws=\x00" as *u8); gn(1, vd[0]); gp(1, "(0=ALIVE) crashed_ws=\x00" as *u8); gn(1, vd[1]); gp(1, "(1=STALLED)\x00" as *u8)
  if live_ok == 1 { gp(1, " PASS\n\x00" as *u8) } else { gp(1, " FAIL\n\x00" as *u8) }
  gp(1, "verdict=\x00" as *u8)
  if green == 1 { gp(1, "GREEN\n\x00" as *u8) } else { gp(1, "RED\n\x00" as *u8) }
223
  // single assembled evidence line -> fa_appendz (dogfoods R0b)
  let rec: *u8 = sys_mmap(RECCAP + 16)
  var o: i64 = 0
  o = fa_cat(rec, o, "WMS-LIVE13 epoch=\x00" as *u8); o = fa_catn(rec, o, sys_now_realtime_sec())
  o = fa_cat(rec, o, " good_torn=\x00" as *u8); o = fa_catn(rec, o, good_torn)
  o = fa_cat(rec, o, " bad_torn=\x00" as *u8); o = fa_catn(rec, o, bad_torn)
  o = fa_cat(rec, o, " live_ok=\x00" as *u8); o = fa_catn(rec, o, live_ok)
  o = fa_cat(rec, o, " verdict=\x00" as *u8)
  if green == 1 { o = fa_cat(rec, o, "GREEN\x00" as *u8) } else { o = fa_cat(rec, o, "RED\x00" as *u8) }
  rec[o] = 0 as u8
  if logfd > 0 { fa_appendz(CLG_LOG, rec, RECCAP); sys_close(logfd) }
235
  if green == 1 { return 0 }
  return 1
238}
nx_conductor_live_wms_gate.nx source

`nx_conductor_live_wms_gate.nx` source
