nx_ws_ledger_gate.nx source
↩ module page · 286 lines · 12969 B
1// nx_ws_ledger_gate.nx -- the REFEREE for WMS-R2 (workstream transition ledger).
2//
3// Proves the nx_ws_ledger reflog (a) survives concurrent writers with ZERO torn
4// entries AND ledger_replay reconstructs the correct final state, (b) FLAGS a
5// corrupted entry instead of silently skipping it, and (c) that the detector can
6// actually SEE failure (the mandatory negative control: the OLD multi-write path
7// tears). No-false-green: GREEN requires all lanes.
8//
9// GOOD lane: NWORKERS fork concurrently; worker w writes NTRANS transitions
10// for ws-id=w via ledger_append (the R0 single-write primitive).
11// The terminal transition sets new = TERMINAL(w) deterministically.
12// Expect: total_lines == NWORKERS*NTRANS, flagged == 0, and
13// state[w] == TERMINAL(w) for every w (replay rebuilt final state).
14// TAMPER: a log of K good records PLUS one deliberately-broken raw record
15// (missing the " END" sentinel) -> ledger_replay MUST flag it
16// (flagged >= 1). flagged == 0 => corruption silently swallowed
17// => RED.
18// NEG-CTRL: same concurrency as GOOD, but each transition emitted the OLD way
19// -- a SEQUENCE of separate sys_write() calls -> interleaving ->
20// torn entries. ledger_replay MUST flag > 0. flagged == 0 => the
21// detector is worthless => RED.
22// BOUND: an oversized record (rec_len+1 > cap) must be REJECTED with -2.
23//
24// Output: rows -> stdout + knowledge/status/ws_ledger_gate.log, then a verdict.
25// Exit 0 GREEN, 1 RED. Sovereign: only nx_syscalls + nx_ws_ledger.
26// license_tier: ORIGINAL
27import "nx_syscalls.nx"
28import "nx_ws_ledger.nx"
29
30const NWORKERS: i64 = 16 // concurrency (matches R0 gate)
31const NTRANS: i64 = 300 // transitions per worker
32const RECCAP: i64 = 256 // bounded record size
33const MAXWS: i64 = 64 // state[] capacity (> NWORKERS)
34const NSTATES: i64 = 5 // 0=TODO 1=WIP 2=VIEW 3=DONE 4=NOVEL
35
36// deterministic terminal state for worker w (what replay must reconstruct).
37func terminal(w: i64) -> i64 { return (w + 3) % NSTATES }
38
39// dual-sink writers (lifted from the R0 gate)
40func gp(logfd: i64, s: *u8) -> i64 {
41 var n: i64 = 0
42 while s[n] != (0 as u8) { n = n + 1 }
43 sys_write(1, s, n)
44 if logfd > 0 { sys_write(logfd, s, n) }
45 return 0
46}
47func gn(logfd: i64, v: i64) -> i64 {
48 let bb: *u8 = sys_mmap(28)
49 var m: i64 = v
50 if m < 0 { sys_write(1, "-\x00" as *u8, 1); if logfd > 0 { sys_write(logfd, "-\x00" as *u8, 1) } m = 0 - m }
51 let t: *u8 = sys_mmap(28)
52 var k: i64 = 0
53 if m == 0 { t[0] = 48 as u8; k = 1 }
54 while m > 0 { t[k] = (48 + (m % 10)) as u8; m = m / 10; k = k + 1 }
55 var i: i64 = 0
56 while i < k { bb[i] = t[k - 1 - i]; i = i + 1 }
57 sys_write(1, bb, k)
58 if logfd > 0 { sys_write(logfd, bb, k) }
59 return 0
60}
61
62// GOOD worker: NTRANS transitions for ws-id=w via the single-write primitive.
63// Each step transitions old=(i%NSTATES) -> new; the FINAL step pins new=terminal(w)
64// so replay's last-writer-wins result is deterministically known.
65func good_worker(w: i64) -> i64 {
66 let pid: i64 = __syscall(39, 0, 0, 0, 0, 0, 0) // getpid (actor)
67 var i: i64 = 0
68 while i < NTRANS {
69 var nw: i64 = (i + 1) % NSTATES
70 if i == NTRANS - 1 { nw = terminal(w) }
71 ledger_append("/tmp/wsl_good.log\x00" as *u8, w, i % NSTATES, nw, pid, i, w * 1000 + i)
72 i = i + 1
73 }
74 return 0
75}
76
77// NEG-CONTROL worker: SAME transition fields, but emitted as a SEQUENCE of
78// separate sys_write() calls (the torn-line root cause). Opens O_APPEND once,
79// then per transition many little writes + newline -> concurrent interleaving.
80func bad_worker(w: i64) -> i64 {
81 let pid: i64 = __syscall(39, 0, 0, 0, 0, 0, 0)
82 let fd: i64 = sys_openat_append("/tmp/wsl_bad.log\x00" as *u8, 0x1a4)
83 if fd < 0 { return 0 - 1 }
84 let nb: *u8 = sys_mmap(28)
85 var i: i64 = 0
86 while i < NTRANS {
87 var nw: i64 = (i + 1) % NSTATES
88 if i == NTRANS - 1 { nw = terminal(w) }
89 sys_write(fd, "WSX ws=\x00" as *u8, 7)
90 // ws
91 var l: i64 = 0; var m: i64 = w
92 if m == 0 { nb[0] = 48 as u8; l = 1 }
93 while m > 0 { nb[l] = (48 + (m % 10)) as u8; m = m / 10; l = l + 1 }
94 let d1: *u8 = sys_mmap(28); var x: i64 = 0
95 while x < l { d1[x] = nb[l - 1 - x]; x = x + 1 }
96 sys_write(fd, d1, l)
97 sys_write(fd, " old=\x00" as *u8, 5)
98 l = 0; m = i % NSTATES
99 if m == 0 { nb[0] = 48 as u8; l = 1 }
100 while m > 0 { nb[l] = (48 + (m % 10)) as u8; m = m / 10; l = l + 1 }
101 let d2: *u8 = sys_mmap(28); x = 0
102 while x < l { d2[x] = nb[l - 1 - x]; x = x + 1 }
103 sys_write(fd, d2, l)
104 sys_write(fd, " new=\x00" as *u8, 5)
105 l = 0; m = nw
106 if m == 0 { nb[0] = 48 as u8; l = 1 }
107 while m > 0 { nb[l] = (48 + (m % 10)) as u8; m = m / 10; l = l + 1 }
108 let d3: *u8 = sys_mmap(28); x = 0
109 while x < l { d3[x] = nb[l - 1 - x]; x = x + 1 }
110 sys_write(fd, d3, l)
111 sys_write(fd, " actor=\x00" as *u8, 7)
112 l = 0; m = pid
113 if m == 0 { nb[0] = 48 as u8; l = 1 }
114 while m > 0 { nb[l] = (48 + (m % 10)) as u8; m = m / 10; l = l + 1 }
115 let d4: *u8 = sys_mmap(28); x = 0
116 while x < l { d4[x] = nb[l - 1 - x]; x = x + 1 }
117 sys_write(fd, d4, l)
118 sys_write(fd, " reason=\x00" as *u8, 8)
119 l = 0; m = i
120 if m == 0 { nb[0] = 48 as u8; l = 1 }
121 while m > 0 { nb[l] = (48 + (m % 10)) as u8; m = m / 10; l = l + 1 }
122 let d5: *u8 = sys_mmap(28); x = 0
123 while x < l { d5[x] = nb[l - 1 - x]; x = x + 1 }
124 sys_write(fd, d5, l)
125 sys_write(fd, " ev=0 epoch=0 END\x00" as *u8, 16)
126 sys_write(fd, "\n\x00" as *u8, 1)
127 i = i + 1
128 }
129 sys_close(fd)
130 return 0
131}
132
133// spawn NWORKERS children running f(w), wait all. which: 0=good, 1=bad.
134func spawn_all(which: i64) -> i64 {
135 let pids: *i64 = sys_mmap(8 * (NWORKERS + 4)) as *i64
136 var w: i64 = 0
137 while w < NWORKERS {
138 let pid: i64 = sys_fork()
139 if pid == 0 {
140 if which == 0 { good_worker(w) }
141 if which == 1 { bad_worker(w) }
142 sys_exit(0)
143 }
144 pids[w] = pid
145 w = w + 1
146 }
147 let st: *i64 = sys_mmap(16) as *i64
148 w = 0
149 while w < NWORKERS {
150 sys_wait4(pids[w], st, 0)
151 w = w + 1
152 }
153 return 0
154}
155
156func main() -> i64 {
157 let logfd: i64 = sys_openat_append("knowledge/status/ws_ledger_gate.log\x00" as *u8, 0x1a4)
158 gp(logfd, "WS-LEDGER epoch=\x00" as *u8)
159 gn(logfd, sys_now_realtime_sec())
160 gp(logfd, " workers=\x00" as *u8); gn(logfd, NWORKERS)
161 gp(logfd, " trans=\x00" as *u8); gn(logfd, NTRANS)
162 gp(logfd, " cap=\x00" as *u8); gn(logfd, RECCAP)
163 gp(logfd, "\n\x00" as *u8)
164
165 // fresh files each run (truncate)
166 let g0: i64 = sys_openat_wr("/tmp/wsl_good.log\x00" as *u8, 0x1a4)
167 if g0 > 0 { sys_close(g0) }
168 let b0: i64 = sys_openat_wr("/tmp/wsl_bad.log\x00" as *u8, 0x1a4)
169 if b0 > 0 { sys_close(b0) }
170 let t0: i64 = sys_openat_wr("/tmp/wsl_tamper.log\x00" as *u8, 0x1a4)
171 if t0 > 0 { sys_close(t0) }
172
173 // ===== GOOD lane: concurrent framed single-write transitions =====
174 spawn_all(0)
175 let state: *i64 = sys_mmap(8 * MAXWS) as *i64
176 var si: i64 = 0
177 while si < MAXWS { state[si] = 0 - 1; si = si + 1 } // init unseen
178 let outs: *i64 = sys_mmap(32) as *i64
179 ledger_replay("/tmp/wsl_good.log\x00" as *u8, state, outs, MAXWS)
180 let good_lines: i64 = outs[0]
181 let good_applied: i64 = outs[1]
182 let good_flagged: i64 = outs[2]
183 let want_lines: i64 = NWORKERS * NTRANS
184 // verify replay reconstructed the correct FINAL state per workstream
185 var state_ok: i64 = 0
186 var w: i64 = 0
187 while w < NWORKERS {
188 if state[w] == terminal(w) { state_ok = state_ok + 1 }
189 w = w + 1
190 }
191
192 // ===== TAMPER: good records + one deliberately-broken raw record =====
193 // write a few clean records via the real capability ...
194 var ki: i64 = 0
195 while ki < 5 {
196 ledger_append("/tmp/wsl_tamper.log\x00" as *u8, 0, 0, 1, 7, ki, ki)
197 ki = ki + 1
198 }
199 // ... then corrupt: one raw record MISSING the " END" sentinel + newline.
200 let tfd: i64 = sys_openat_append("/tmp/wsl_tamper.log\x00" as *u8, 0x1a4)
201 if tfd > 0 {
202 sys_write(tfd, "WSX ws=0 old=1 new=2 actor=7 reason=9 ev=0 epoch=0 BROKEN\n\x00" as *u8, 58)
203 sys_close(tfd)
204 }
205 let tstate: *i64 = sys_mmap(8 * MAXWS) as *i64
206 si = 0
207 while si < MAXWS { tstate[si] = 0 - 1; si = si + 1 }
208 let touts: *i64 = sys_mmap(32) as *i64
209 ledger_replay("/tmp/wsl_tamper.log\x00" as *u8, tstate, touts, MAXWS)
210 let tamper_flagged: i64 = touts[2]
211
212 // ===== NEGATIVE CONTROL: old multi-write path under concurrency =====
213 spawn_all(1)
214 let bstate: *i64 = sys_mmap(8 * MAXWS) as *i64
215 si = 0
216 while si < MAXWS { bstate[si] = 0 - 1; si = si + 1 }
217 let bouts: *i64 = sys_mmap(32) as *i64
218 ledger_replay("/tmp/wsl_bad.log\x00" as *u8, bstate, bouts, MAXWS)
219 let bad_lines: i64 = bouts[0]
220 let bad_flagged: i64 = bouts[2]
221
222 // ===== BOUND: oversized record must be REJECTED with -2 =====
223 // a big record into a tiny cap must be rejected (never torn) by the primitive.
224 let trec: *u8 = sys_mmap(64)
225 var ai: i64 = 0
226 while ai < 32 { trec[ai] = 65 as u8; ai = ai + 1 }
227 let bound_rc: i64 = fa_append("/tmp/wsl_void.log\x00" as *u8, trec, 32, 8)
228
229 // ===== rows =====
230 gp(logfd, "WSL row=good_lines got=\x00" as *u8); gn(logfd, good_lines)
231 gp(logfd, " want=\x00" as *u8); gn(logfd, want_lines)
232 if good_lines == want_lines { gp(logfd, " verdict=PASS\n\x00" as *u8) } else { gp(logfd, " verdict=FAIL\n\x00" as *u8) }
233
234 gp(logfd, "WSL row=good_flagged got=\x00" as *u8); gn(logfd, good_flagged)
235 gp(logfd, " want=0\x00" as *u8)
236 if good_flagged == 0 { gp(logfd, " verdict=PASS\n\x00" as *u8) } else { gp(logfd, " verdict=FAIL\n\x00" as *u8) }
237
238 gp(logfd, "WSL row=replay_state_ok got=\x00" as *u8); gn(logfd, state_ok)
239 gp(logfd, " want=\x00" as *u8); gn(logfd, NWORKERS)
240 if state_ok == NWORKERS { gp(logfd, " verdict=PASS\n\x00" as *u8) } else { gp(logfd, " verdict=FAIL\n\x00" as *u8) }
241
242 gp(logfd, "WSL row=tamper_flagged got=\x00" as *u8); gn(logfd, tamper_flagged)
243 gp(logfd, " want>=1\x00" as *u8)
244 if tamper_flagged >= 1 { gp(logfd, " verdict=PASS(corruption-surfaced)\n\x00" as *u8) } else { gp(logfd, " verdict=FAIL(corruption-swallowed)\n\x00" as *u8) }
245
246 gp(logfd, "WSL row=neg_control bad_lines=\x00" as *u8); gn(logfd, bad_lines)
247 gp(logfd, " bad_flagged=\x00" as *u8); gn(logfd, bad_flagged)
248 gp(logfd, " want=POSITIVE\x00" as *u8)
249 if bad_flagged > 0 { gp(logfd, " verdict=PASS(detector-sees-tearing)\n\x00" as *u8) } else { gp(logfd, " verdict=FAIL(no-tearing-detected)\n\x00" as *u8) }
250
251 gp(logfd, "WSL row=bound_oversize got=\x00" as *u8); gn(logfd, bound_rc)
252 gp(logfd, " want=-2\x00" as *u8)
253 if bound_rc == (0 - 2) { gp(logfd, " verdict=PASS\n\x00" as *u8) } else { gp(logfd, " verdict=FAIL\n\x00" as *u8) }
254
255 // ===== verdict (no false green) =====
256 var green: i64 = 1
257 if good_lines != want_lines { green = 0 } // nothing lost
258 if good_flagged != 0 { green = 0 } // good lane zero torn
259 if state_ok != NWORKERS { green = 0 } // replay rebuilt correct final state
260 if tamper_flagged < 1 { green = 0 } // corruption SURFACED
261 if bad_flagged <= 0 { green = 0 } // NEG-CONTROL actually tears
262 if bound_rc != (0 - 2) { green = 0 } // bound enforced
263
264 gp(logfd, "WS-LEDGER verdict=\x00" as *u8)
265 if green == 1 { gp(logfd, "GREEN\x00" as *u8) } else { gp(logfd, "RED\x00" as *u8) }
266 gp(logfd, " good_lines=\x00" as *u8); gn(logfd, good_lines)
267 gp(logfd, " good_applied=\x00" as *u8); gn(logfd, good_applied)
268 gp(logfd, " good_flagged=\x00" as *u8); gn(logfd, good_flagged)
269 gp(logfd, " state_ok=\x00" as *u8); gn(logfd, state_ok)
270 gp(logfd, " tamper_flagged=\x00" as *u8); gn(logfd, tamper_flagged)
271 gp(logfd, " bad_flagged=\x00" as *u8); gn(logfd, bad_flagged)
272 gp(logfd, " bound=\x00" as *u8); gn(logfd, bound_rc)
273 if green == 0 {
274 gp(logfd, " reason=\x00" as *u8)
275 if good_lines != want_lines { gp(logfd, "good-lines-lost \x00" as *u8) }
276 if good_flagged != 0 { gp(logfd, "GOOD-TORE \x00" as *u8) }
277 if state_ok != NWORKERS { gp(logfd, "replay-wrong-state \x00" as *u8) }
278 if tamper_flagged < 1 { gp(logfd, "tamper-swallowed \x00" as *u8) }
279 if bad_flagged <= 0 { gp(logfd, "neg-control-saw-no-tearing \x00" as *u8) }
280 if bound_rc != (0 - 2) { gp(logfd, "oversize-not-rejected \x00" as *u8) }
281 }
282 gp(logfd, "\n\x00" as *u8)
283 if logfd > 0 { sys_close(logfd) }
284 if green == 1 { return 0 }
285 return 1
286}