nx_ws_ledger.nx source
↩ module page · 204 lines · 8142 B
1// nx_ws_ledger.nx -- WMS-R2: the workstream TRANSITION LEDGER (reflog).
2//
3// module: nishi-core.wms.ws_ledger
4// capability: CORE_COMPUTE (append-only transition ledger built ON WMS-R0)
5//
6// WHAT THIS CLOSES: a workstream's life is a sequence of state changes
7// (TODO->WIP->VIEW->DONE ...). We need an append-only LEDGER (a reflog) that
8// records EVERY transition durably AND lets us REPLAY the log to rebuild the
9// current state of every workstream. Two correctness hazards:
10// 1) concurrent writers tearing the log (the WMS-R0 root cause) -- defeated
11// here by emitting each transition as ONE framed record via fa_append.
12// 2) a corrupt/torn entry being SILENTLY skipped during replay -- defeated
13// here by FLAGGING (counting) any malformed line instead of dropping it.
14//
15// REUSE / lineage:
16// - fa_append / fa_cat / fa_catn (nx_framed_append.nx, WMS-R0): the atomic
17// single-write framing primitive -> each transition = exactly one write().
18// - the byte-scan line parser + token matcher idiom is lifted from
19// nx_reconcile.nx / nx_framed_append_gate.count_torn.
20// - sys_read_file (nx_syscalls.nx): whole-file replay read.
21// Sovereign: only nx_syscalls + nx_framed_append. Additive: no organ modified.
22// Additive-only data (Cardinal 13): the ledger is append-only; ledger_append
23// NEVER mutates prior bytes, history is sacred -> replay/rollback/audit work.
24// license_tier: ORIGINAL
25import "nx_syscalls.nx"
26import "nx_framed_append.nx"
27
28const WSL_RECCAP: i64 = 256 // bounded record size (matches R0 RECCAP)
29
30// Record shape (ONE framed line, NO embedded '\n'):
31// "WSX ws=<id> old=<s> new=<s> actor=<a> reason=<r> ev=<cid> epoch=<sec> END"
32// Leading "WSX " sentinel + trailing " END" sentinel are the well-formedness
33// anchors replay/tamper detection keys on (twin of R0's "FA w=" / " END").
34
35// Assemble one transition record into buf (NO trailing newline); returns rec_len.
36// Mirrors build_rec in the R0 gate (fa_cat / fa_catn assembly).
37func wsl_build(buf: *u8, ws: i64, old_st: i64, new_st: i64,
38 actor: i64, reason: i64, ev: i64, epoch: i64) -> i64 {
39 var o: i64 = 0
40 o = fa_cat(buf, o, "WSX ws=\x00" as *u8)
41 o = fa_catn(buf, o, ws)
42 o = fa_cat(buf, o, " old=\x00" as *u8)
43 o = fa_catn(buf, o, old_st)
44 o = fa_cat(buf, o, " new=\x00" as *u8)
45 o = fa_catn(buf, o, new_st)
46 o = fa_cat(buf, o, " actor=\x00" as *u8)
47 o = fa_catn(buf, o, actor)
48 o = fa_cat(buf, o, " reason=\x00" as *u8)
49 o = fa_catn(buf, o, reason)
50 o = fa_cat(buf, o, " ev=\x00" as *u8)
51 o = fa_catn(buf, o, ev)
52 o = fa_cat(buf, o, " epoch=\x00" as *u8)
53 o = fa_catn(buf, o, epoch)
54 o = fa_cat(buf, o, " END\x00" as *u8)
55 return o
56}
57
58// THE WRITE CAPABILITY. Stamp the wall-clock epoch, assemble the record, append
59// it via the R0 single-write primitive. Returns fa_append's rc:
60// > 0 bytes written (success)
61// -1 open failure -2 oversized (rejected, never torn) -3 short write
62// Append-only / idempotent-safe: never mutates prior bytes (Cardinal 13).
63func ledger_append(path: *u8, ws: i64, old_st: i64, new_st: i64,
64 actor: i64, reason: i64, ev: i64) -> i64 {
65 let epoch: i64 = sys_now_realtime_sec()
66 let buf: *u8 = sys_mmap(WSL_RECCAP + 16)
67 let rl: i64 = wsl_build(buf, ws, old_st, new_st, actor, reason, ev, epoch)
68 return fa_append(path, buf, rl, WSL_RECCAP)
69}
70
71// Token matcher: does the `pat` (plen bytes) occur at b[pos..pos+plen) within
72// the line [ls,le)? (reconcile rc_at idiom.) Returns 1/0.
73func wsl_at(b: *u8, pos: i64, le: i64, pat: *u8, plen: i64) -> i64 {
74 if pos + plen > le { return 0 }
75 var k: i64 = 0
76 while k < plen {
77 if b[pos + k] != pat[k] { return 0 }
78 k = k + 1
79 }
80 return 1
81}
82
83// Parse the decimal integer that starts at b[pos] (stops at first non-digit).
84// Returns the value; *adv set to the index just past the last digit.
85func wsl_parse_int(b: *u8, pos: i64, le: i64, adv: *i64) -> i64 {
86 var v: i64 = 0
87 var p: i64 = pos
88 while p < le {
89 let c: i64 = b[p] as i64
90 if c < 48 { p = le }
91 if c >= 48 {
92 if c > 57 { p = le }
93 if c <= 57 { v = v * 10 + (c - 48); p = p + 1 }
94 }
95 }
96 *adv = p
97 return v
98}
99
100// Find the value of "<key>" within line [ls,le): scans for the key token, then
101// parses the integer after it. Returns the value, or -1 if the key is absent.
102func wsl_field(b: *u8, ls: i64, le: i64, key: *u8, klen: i64) -> i64 {
103 var p: i64 = ls
104 while p < le {
105 if wsl_at(b, p, le, key, klen) == 1 {
106 let adv: *i64 = sys_mmap(16) as *i64
107 let v: i64 = wsl_parse_int(b, p + klen, le, adv)
108 return v
109 }
110 p = p + 1
111 }
112 return 0 - 1
113}
114
115// Well-formedness of line [ls,le) (le = index of the '\n'), identical 3-anchor
116// discipline to the R0 gate's count_torn:
117// - begins "WSX "
118// - the 4 bytes immediately before '\n' are " END"
119// - exactly one " END" token in the line
120// - no SECOND "WSX " record-head embedded after ls (interleaving sentinel)
121// Returns 1 if well-formed, 0 if TORN/CORRUPT.
122func wsl_wellformed(b: *u8, ls: i64, le: i64) -> i64 {
123 let llen: i64 = le - ls
124 if llen < 8 { return 0 }
125 // prefix "WSX " (4 bytes)
126 if b[ls] != (87 as u8) { return 0 } // W
127 if b[ls + 1] != (83 as u8) { return 0 } // S
128 if b[ls + 2] != (88 as u8) { return 0 } // X
129 if b[ls + 3] != (32 as u8) { return 0 } // space
130 // suffix " END" immediately before '\n'
131 let e: i64 = le - 1
132 if b[e - 3] != (32 as u8) { return 0 } // space
133 if b[e - 2] != (69 as u8) { return 0 } // E
134 if b[e - 1] != (78 as u8) { return 0 } // N
135 if b[e] != (68 as u8) { return 0 } // D
136 // exactly one " END" token inside the line
137 var ends: i64 = 0
138 var j: i64 = ls
139 while j + 3 < le + 1 {
140 if b[j] == (32 as u8) {
141 if b[j + 1] == (69 as u8) {
142 if b[j + 2] == (78 as u8) {
143 if b[j + 3] == (68 as u8) { ends = ends + 1 }
144 }
145 }
146 }
147 j = j + 1
148 }
149 if ends != 1 { return 0 }
150 // no second "WSX " record-head embedded (head must appear only at ls)
151 var k: i64 = ls + 1
152 while k + 3 < le + 1 {
153 if b[k] == (87 as u8) {
154 if b[k + 1] == (83 as u8) {
155 if b[k + 2] == (88 as u8) {
156 if b[k + 3] == (32 as u8) { return 0 }
157 }
158 }
159 }
160 k = k + 1
161 }
162 return 1
163}
164
165// THE REPLAY CAPABILITY. Read the whole log, scan line-by-line. For each
166// WELL-FORMED record set state[ws] = new_st (last-writer-wins == final state).
167// A malformed/torn/corrupt line is FLAGGED (counted), NOT silently skipped.
168// state[] : caller-mmapped i64[maxws], MUST be pre-init to -1 (unseen).
169// outs[0] = total_lines outs[1] = applied outs[2] = flagged
170// `maxws` bounds the ws-id range state[] can hold (out-of-range ws still counts
171// as applied at the parse level but is not stored). Returns flagged count.
172func ledger_replay(path: *u8, state: *i64, outs: *i64, maxws: i64) -> i64 {
173 let szp: *i64 = sys_mmap(16) as *i64
174 let b: *u8 = sys_read_file(path, szp)
175 let sz: i64 = szp[0]
176 var lines: i64 = 0
177 var applied: i64 = 0
178 var flagged: i64 = 0
179 var ls: i64 = 0
180 var i: i64 = 0
181 while i < sz {
182 if b[i] == (10 as u8) {
183 lines = lines + 1
184 if wsl_wellformed(b, ls, i) == 1 {
185 let ws: i64 = wsl_field(b, ls, i, "ws=\x00" as *u8, 3)
186 let nw: i64 = wsl_field(b, ls, i, "new=\x00" as *u8, 4)
187 if ws >= 0 {
188 if nw >= 0 {
189 if ws < maxws { state[ws] = nw }
190 applied = applied + 1
191 } else { flagged = flagged + 1 }
192 } else { flagged = flagged + 1 }
193 } else {
194 flagged = flagged + 1
195 }
196 ls = i + 1
197 }
198 i = i + 1
199 }
200 outs[0] = lines
201 outs[1] = applied
202 outs[2] = flagged
203 return flagged
204}