code wiki / _hdl_build / nx_workstream_store.nx
nx_workstream_store.nx source
↩ module page · 254 lines · 11514 B
1// nx_workstream_store.nx -- WMS-R1 LIB: the content-addressed WORKSTREAM REGISTRY (SSOT),
2// the team's single machine-readable enumeration of EVERY workstream across the 12 empires.
3// Built ON the sovereign segment store (knowledge/store/ws-*, via nx_seg_store) -- NO TSV, NO
4// SQL, NO git. The structural twin of nx_known_issue_store: that catalogues "have we SEEN this
5// failure?"; THIS catalogues "what WORKSTREAMS exist, and is the registry COMPLETE?". It closes
6// the lost-things bug (crash recovery missed entire empires because no SSOT enumerated them).
7//
8// Record schema (all keys under prefix knowledge/store/ws-), TAB-separated value fields:
9// ws:ids -> index: TAB-separated list of every workstream id (the ws_manifest walk order)
10// ws:empires -> index: TAB-separated list of the 12 empire ids (enables empire-completeness)
11// ws:<id> -> id <tab> empire <tab> state <tab> last_touched <tab> memory_link
12// <tab> code_link <tab> deps
13// state = ACTIVE|DONE|BLOCKED|FLOATING|ABSENT (data-driven label set)
14// last_touched = epoch seconds (decimal) or 0 when unknown
15// memory_link = memory file basename or "-"
16// code_link = organ/file path or "-"
17// deps = comma-joined dep ids or "-"
18// ws:provenance -> authorship breadcrumb (verified-vs-staging is machine-visible in the store)
19//
20// REUSE / lineage: ss_* (atomic tmp+fsync+rename commit point) is the substrate; the ws:ids /
21// ws_field TAB-parser / ws_seg_next idioms are lifted straight from nx_known_issue_store
22// (ki:ids / ki_field / ki_seg_next_p) -- the registry is DATA-driven, adding a stream is a seed
23// edit, not new code. Sovereign: imports only nx_seg_store + nx_syscalls (no gcc). license_tier: ORIGINAL
24import "nx_seg_store.nx"
25import "nx_syscalls.nx"
26
27const WS_PREFIX: *u8 = "knowledge/store/ws-"
28
29// get/audit verdict codes (data-driven; no magic numbers)
30const WS_FOUND: i64 = 1 // record retrievable
31const WS_GONE: i64 = 0 // tombstoned
32const WS_UNKNOWN: i64 = 0 - 1 // never registered -- the NEG-CONTROL honest answer
33
34func ws_len(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } return n }
35
36// latest record for `key` under `prefix`: 1=found (ptr/len set), 0=tombstoned, -1=absent.
37// Thin over ss_get so the NEG-CONTROL (never-registered id -> -1) is the store's honest answer,
38// never a fabricated hit.
39func ws_get_p(prefix: *u8, key: *u8, ptrout: *i64, lenout: *i64) -> i64 {
40 return ss_get(prefix, key, ptrout, lenout)
41}
42func ws_get(key: *u8, ptrout: *i64, lenout: *i64) -> i64 {
43 return ss_get(WS_PREFIX, key, ptrout, lenout)
44}
45
46// next segment id for a commit under `prefix` = fresh max+1 (ss_next_segid, UNCAPPED). 2026-07-16 fix
47// of the legacy-cap clobber class: 1 + capped ss_manifest count stuck at the cap forever once exceeded,
48// so every later WMS commit overwrote the same segment (SSOT registration loss -- the reg_put bug class).
49// Empty store stays 1-based (mirrors ki_seg_next_p's convention).
50func ws_seg_next(prefix: *u8) -> i64 {
51 let n: i64 = ss_next_segid(prefix)
52 if n < 1 { return 1 }
53 return n
54}
55
56// extract the f-th (0-based) TAB-separated field of rec[0..rlen) into out (NUL-terminated);
57// returns the field length. ONE parser for empire/state/links by construction (mirrors ki_field).
58func ws_field(rec: *u8, rlen: i64, f: i64, out: *u8) -> i64 {
59 var cur: i64 = 0
60 var i: i64 = 0
61 var o: i64 = 0
62 while cur < f {
63 if i >= rlen { out[0] = 0 as u8; return 0 }
64 if rec[i] == (9 as u8) { cur = cur + 1 }
65 i = i + 1
66 }
67 var go: i64 = 1
68 while go == 1 {
69 if i >= rlen { go = 0 } else {
70 if rec[i] == (9 as u8) { go = 0 } else { out[o] = rec[i]; o = o + 1; i = i + 1 }
71 }
72 }
73 out[o] = 0 as u8
74 return o
75}
76
77// is NUL-terminated `tok` a TAB-bounded member of list[0..llen)? (empire-membership check)
78// Matches a whole TAB-separated token, not a substring -- "E-CORE" must not match "E-CORE2".
79func ws_member(list: *u8, llen: i64, tok: *u8) -> i64 {
80 let tl: i64 = ws_len(tok)
81 var i: i64 = 0
82 while i < llen {
83 // start of a field iff i==0 or the previous byte is a TAB
84 var atstart: i64 = 0
85 if i == 0 { atstart = 1 }
86 if i > 0 { if list[i - 1] == (9 as u8) { atstart = 1 } }
87 if atstart == 1 {
88 // field spans [i .. fe) up to next TAB or end of list
89 var fe: i64 = i
90 var go: i64 = 1
91 while go == 1 {
92 if fe >= llen { go = 0 } else {
93 if list[fe] == (9 as u8) { go = 0 } else { fe = fe + 1 }
94 }
95 }
96 if fe - i == tl {
97 var eq: i64 = 1
98 var j: i64 = 0
99 while j < tl { if list[i + j] != tok[j] { eq = 0; j = tl } else { j = j + 1 } }
100 if eq == 1 { return 1 }
101 }
102 }
103 i = i + 1
104 }
105 return 0
106}
107
108// ws_put: register/update ONE workstream record (idempotent: skip-if-unchanged via a current-value
109// byte-compare; additive new version otherwise). `val` is the prebuilt TAB record. Returns
110// 0 committed a new version
111// 1 unchanged -- skipped (idempotent re-run)
112// <0 commit error (negative ss_commit rc)
113func ws_put_p(prefix: *u8, key: *u8, val: *u8) -> i64 {
114 let pq: *i64 = sys_mmap(16) as *i64
115 let lq: *i64 = sys_mmap(16) as *i64
116 let vl: i64 = ws_len(val)
117 if ss_get(prefix, key, pq, lq) == 1 {
118 if lq[0] == vl {
119 let b: *u8 = pq[0] as *u8
120 var same: i64 = 1
121 var i: i64 = 0
122 while i < vl { if b[i] != val[i] { same = 0; i = vl } else { i = i + 1 } }
123 if same == 1 { return 1 }
124 }
125 }
126 let w: *i64 = ss_begin()
127 ss_add(w, 1, key, val, vl)
128 let segid: i64 = ws_seg_next(prefix)
129 return ss_commit(prefix, w, segid)
130}
131func ws_put(key: *u8, val: *u8) -> i64 {
132 // WMS-R8 debt eaten 2026-07-18: the DEFAULT entry is LOCKED (flock EX on <prefix>wlock, the
133 // proven ws_put_locked frame) -- every casual caller was racing the segid/manifest RMW.
134 // ws_put_p stays the RAW primitive for explicit single-writer recovery + gate neg-controls.
135 // Lock-open failure degrades to the raw path (some filesystems lack flock; tmp+rename remains).
136 let lp: *u8 = sys_mmap(512)
137 let pfx: *u8 = WS_PREFIX
138 var i: i64 = 0
139 while pfx[i] != (0 as u8) { lp[i] = pfx[i]; i = i + 1 }
140 let suf: *u8 = "wlock" as *u8
141 var j: i64 = 0
142 while suf[j] != (0 as u8) { lp[i] = suf[j]; i = i + 1; j = j + 1 }
143 lp[i] = 0 as u8
144 let lfd: i64 = sys_openat_append(lp, 0x1a4)
145 if lfd < 0 { return ws_put_p(WS_PREFIX, key, val) }
146 sys_flock(lfd, SYS_LOCK_EX)
147 let rc: i64 = ws_put_p(WS_PREFIX, key, val)
148 sys_flock(lfd, SYS_LOCK_UN)
149 sys_close(lfd)
150 return rc
151}
152
153// ws_manifest: enumerate EVERY registered id from ws:ids into ids_out[] (i64 ptrs to NUL-term id
154// strings, each mmap'd). Returns count, or -1 if ws:ids absent. This is the SSOT enumeration --
155// the TAB-walk over ws:ids is the exact ki:ids walk shape (proven idiom).
156func ws_manifest_p(prefix: *u8, ids_out: *i64, cap: i64) -> i64 {
157 let pq: *i64 = sys_mmap(16) as *i64
158 let lq: *i64 = sys_mmap(16) as *i64
159 if ss_get(prefix, "ws:ids" as *u8, pq, lq) != 1 { return 0 - 1 }
160 let ids: *u8 = pq[0] as *u8
161 let idn: i64 = lq[0]
162 var cnt: i64 = 0
163 var i: i64 = 0
164 while i < idn {
165 let idbuf: *u8 = sys_mmap(128)
166 var o: i64 = 0
167 var go: i64 = 1
168 while go == 1 {
169 if i >= idn { go = 0 } else {
170 if ids[i] == (9 as u8) { i = i + 1; go = 0 } else { idbuf[o] = ids[i]; o = o + 1; i = i + 1 }
171 }
172 }
173 idbuf[o] = 0 as u8
174 if o > 0 { if cnt < cap { ids_out[cnt] = idbuf as i64; cnt = cnt + 1 } }
175 }
176 return cnt
177}
178func ws_manifest(ids_out: *i64, cap: i64) -> i64 {
179 return ws_manifest_p(WS_PREFIX, ids_out, cap)
180}
181
182// ws_audit_complete: for EVERY id in ws:ids, the ws:<id> segment MUST be retrievable (ss_get==1)
183// AND its empire field (field 1) MUST be a member of ws:empires. Returns the count of MISSING/orphan
184// streams (0 == complete/GREEN); fills missing_out[] with the failing ids (LOUD reporting). This is
185// what turns a removed/torn/never-committed segment into a reported MISSING, never a silent gap.
186func ws_audit_complete_p(prefix: *u8, missing_out: *i64, cap: i64) -> i64 {
187 let pq: *i64 = sys_mmap(16) as *i64
188 let lq: *i64 = sys_mmap(16) as *i64
189 // SCALE FIX: snapshot the whole registry ONCE via ss_open; all lookups below use ss_hget
190 // (in-memory, zero per-call file IO). The old per-key ss_get path mmaps ~100KB/call and
191 // starves the bump-mmap at ~1162 records (SIGTERM/exit 15). Mirrors pq_streq_h in
192 // nx_workstream_popqueue.nx (the populate path already proven at this scale).
193 // ss_open_cached (seq905/962 class fix, seq1347 migration, 2026-07-30). The snapshot-once intent
194 // above is preserved exactly -- this additionally stops the SNAPSHOT ITSELF from leaking the whole
195 // store on every audit call. SAFE: every lookup below is an ss_hget against THIS handle; the store
196 // is never re-entered while it is held.
197 let h: *i64 = ss_open_cached(prefix)
198 if ss_hget(h, "ws:ids" as *u8, pq, lq) != 1 { return 0 - 1 }
199 let ids: *u8 = pq[0] as *u8
200 let idn: i64 = lq[0]
201 // load the empire membership list once
202 let eq: *i64 = sys_mmap(16) as *i64
203 let el: *i64 = sys_mmap(16) as *i64
204 var emp: *u8 = "" as *u8
205 var empn: i64 = 0
206 if ss_hget(h, "ws:empires" as *u8, eq, el) == 1 { emp = eq[0] as *u8; empn = el[0] }
207 let rq: *i64 = sys_mmap(16) as *i64
208 let rl: *i64 = sys_mmap(16) as *i64
209 let keybuf: *u8 = sys_mmap(128)
210 let empbuf: *u8 = sys_mmap(128)
211 var missing: i64 = 0
212 var i: i64 = 0
213 while i < idn {
214 let idbuf: *u8 = sys_mmap(128)
215 var o: i64 = 0
216 var go: i64 = 1
217 while go == 1 {
218 if i >= idn { go = 0 } else {
219 if ids[i] == (9 as u8) { i = i + 1; go = 0 } else { idbuf[o] = ids[i]; o = o + 1; i = i + 1 }
220 }
221 }
222 idbuf[o] = 0 as u8
223 if o > 0 {
224 // key = "ws:" + id
225 keybuf[0] = 119 as u8 // 'w'
226 keybuf[1] = 115 as u8 // 's'
227 keybuf[2] = 58 as u8 // ':'
228 var t: i64 = 0
229 while t < o { keybuf[3 + t] = idbuf[t]; t = t + 1 }
230 keybuf[3 + o] = 0 as u8
231 var bad: i64 = 0
232 if ss_hget(h, keybuf, rq, rl) != 1 {
233 bad = 1 // id in the index but its segment is NOT retrievable -> MISSING
234 } else {
235 // segment present: its empire field must be a real empire id
236 ws_field(rq[0] as *u8, rl[0], 1, empbuf)
237 if ws_member(emp, empn, empbuf) == 0 { bad = 1 } // orphan empire
238 }
239 if bad == 1 {
240 if missing < cap {
241 let mb: *u8 = sys_mmap(128)
242 var z: i64 = 0
243 while z <= o { mb[z] = idbuf[z]; z = z + 1 }
244 missing_out[missing] = mb as i64
245 }
246 missing = missing + 1
247 }
248 }
249 }
250 return missing
251}
252func ws_audit_complete(missing_out: *i64, cap: i64) -> i64 {
253 return ws_audit_complete_p(WS_PREFIX, missing_out, cap)
254}