code wiki / _hdl_build / nx_role_store.nx
nx_role_store.nx source
↩ module page · 182 lines · 7251 B
1// nx_role_store.nx -- LIB: the role-coordination layer in the SOVEREIGN seg-store (knowledge/store/roles-*),
2// replacing the knowledge/roles/*.tsv files. Operator 2026-06-22: "migrate and get rid of these tsvs just nishi
3// ecosystem." NO TSV. Same append-only-event-log + mutable-status-row shape as the .tsv queues, but the bytes
4// live as content-addressed records in the crash-safe seg-store (nx_seg_store), NOT a flat .tsv file.
5//
6// CHANNELS (a channel = a logical .tsv replaced):
7// pmcoord <- knowledge/roles/pm_coord.tsv (PM routing audit log, append-only)
8// <role>q <- knowledge/roles/<role>_queue.tsv (per-role intake queue; row status is MUTABLE: PENDING->APPLIED)
9// <role>l <- knowledge/roles/<role>_ledger.tsv (per-role apply ledger, append-only)
10//
11// RECORD SCHEMA inside a channel (mirrors nx_reader_census_store's keyed model):
12// <ch>:n -> the channel row COUNT (decimal). rewritten each append (last-wins).
13// <ch>:<seq> -> one row, 0-based seq. VALUE = the exact TAB-joined row bytes (the old .tsv line MINUS the
14// trailing newline). A row's CURRENT status = the LATEST version of its <ch>:<seq> key
15// (seg-store last-wins) -- so rq_pump flips PENDING->APPLIED by ss_add-ing a new version,
16// no file rewrite.
17//
18// CONCURRENCY: every mutation holds an EXCLUSIVE flock on RS_LOCK across the read-count -> commit window
19// (mirrors nx_ws_cas / pms_register's wlock) so concurrent submits/records never collide on the segid.
20// HONEST BOUND (no silent cap, rule #11/#21): each commit adds ONE segment; ss_manifest + ss_scan cap at 256.
21// rs_append REFUSES LOUD (returns -2) once the live segment count reaches RS_SEG_MAX, BEFORE staleness can occur.
22// Compaction (ss_compact folds segments->1) is the next rung to lift the bound; it is NOT auto-driven here
23// because the 1+count segid scheme would reuse archived segment ids -- a monotonic-id rung must land first.
24// license_tier: ORIGINAL
25import "nx_seg_store.nx"
26import "nx_syscalls.nx"
27
28const RS_PREFIX: *u8 = "knowledge/store/roles-"
29const RS_LOCK: *u8 = "knowledge/store/roles-wlock"
30const RS_SEG_MAX: i64 = 240
31
32// decimal of v into dst (NUL not written); returns bytes written.
33func rs_itoa(dst: *u8, v: i64) -> i64 {
34 var m: i64 = v
35 var o: i64 = 0
36 if m < 0 { dst[0] = 45 as u8; o = 1; m = 0 - m }
37 let t: *u8 = sys_mmap(28)
38 var k: i64 = 0
39 if m == 0 { t[0] = 48 as u8; k = 1 }
40 while m > 0 { t[k] = (48 + (m % 10)) as u8; m = m / 10; k = k + 1 }
41 var i: i64 = 0
42 while i < k { dst[o + i] = t[k - 1 - i]; i = i + 1 }
43 return o + k
44}
45
46// parse the leading decimal of b[0..n) (ignores any non-digit tail); returns value.
47func rs_atoi(b: *u8, n: i64) -> i64 {
48 var v: i64 = 0
49 var i: i64 = 0
50 var go: i64 = 1
51 while i < n {
52 if go == 1 {
53 let c: i64 = b[i] as i64
54 if c >= 48 { if c <= 57 { v = v * 10 + (c - 48) } else { go = 0 } } else { go = 0 }
55 }
56 i = i + 1
57 }
58 return v
59}
60
61// extract the f-th (0-based) TAB-separated field of rec[0..rlen) into out (NUL-terminated); returns field length.
62func rs_field(rec: *u8, rlen: i64, f: i64, out: *u8) -> i64 {
63 var cur: i64 = 0
64 var i: i64 = 0
65 var o: i64 = 0
66 while cur < f {
67 if i >= rlen { out[0] = 0 as u8; return 0 }
68 if rec[i] == (9 as u8) { cur = cur + 1 }
69 i = i + 1
70 }
71 var go: i64 = 1
72 while go == 1 {
73 if i >= rlen { go = 0 } else {
74 if rec[i] == (9 as u8) { go = 0 } else { out[o] = rec[i]; o = o + 1; i = i + 1 }
75 }
76 }
77 out[o] = 0 as u8
78 return o
79}
80
81// "<ch>:<seq>" into out (NUL-terminated)
82func rs_key(ch: *u8, seq: i64, out: *u8) -> i64 {
83 var o: i64 = ss_cat(out, 0, ch)
84 out[o] = 58 as u8; o = o + 1 // ':'
85 o = ss_catn(out, o, seq)
86 out[o] = 0 as u8
87 return o
88}
89
90// "<ch>:n" into out (the count key)
91func rs_nkey(ch: *u8, out: *u8) -> i64 {
92 var o: i64 = ss_cat(out, 0, ch)
93 out[o] = 58 as u8; o = o + 1 // ':'
94 out[o] = 110 as u8; o = o + 1 // 'n'
95 out[o] = 0 as u8
96 return o
97}
98
99func rs_lock() -> i64 {
100 let fd: i64 = sys_openat_append(RS_LOCK, 0x1a4)
101 if fd >= 0 { sys_flock(fd, SYS_LOCK_EX) }
102 return fd
103}
104
105func rs_unlock(fd: i64) -> i64 {
106 if fd >= 0 { sys_flock(fd, SYS_LOCK_UN); sys_close(fd) }
107 return 0
108}
109
110// live segment count of the role store
111func rs_segcount() -> i64 {
112 let segs: *i64 = sys_mmap(8 * 260) as *i64
113 let n: i64 = ss_manifest(RS_PREFIX, segs)
114 if n < 0 { return 0 }
115 return n
116}
117
118// next segment id for a commit = 1 + current live segment count (monotonic while no compaction)
119func rs_seg_next() -> i64 { return 1 + rs_segcount() }
120
121// current row count of a channel (0 if the channel has no rows yet)
122func rs_count(ch: *u8) -> i64 {
123 let nk: *u8 = sys_mmap(128); rs_nkey(ch, nk)
124 let pq: *i64 = sys_mmap(16) as *i64
125 let lq: *i64 = sys_mmap(16) as *i64
126 if ss_get(RS_PREFIX, nk, pq, lq) != 1 { return 0 }
127 return rs_atoi(pq[0] as *u8, lq[0])
128}
129
130// read row <ch>:<seq> -> ptrout/lenout. 1=found, 0=tombstoned, -1=absent (ss_get semantics).
131func rs_get_seq(ch: *u8, seq: i64, ptrout: *i64, lenout: *i64) -> i64 {
132 let rk: *u8 = sys_mmap(128); rs_key(ch, seq, rk)
133 return ss_get(RS_PREFIX, rk, ptrout, lenout)
134}
135
136// APPEND one row to a channel: writes <ch>:<seq>=val AND bumps <ch>:n in ONE locked commit.
137// Returns the new row's seq (>=0), or -1 commit-failed, or -2 segment-cap-reached (LOUD refusal).
138func rs_append(ch: *u8, val: *u8, vlen: i64) -> i64 {
139 let lk: i64 = rs_lock()
140 if rs_segcount() >= RS_SEG_MAX {
141 rs_unlock(lk)
142 sys_write(2, "nx_role_store: REFUSED append -- roles store at segment cap; compaction rung needed\n" as *u8, 82)
143 return 0 - 2
144 }
145 let n: i64 = rs_count(ch)
146 let rk: *u8 = sys_mmap(128); rs_key(ch, n, rk)
147 let nk: *u8 = sys_mmap(128); rs_nkey(ch, nk)
148 let cb: *u8 = sys_mmap(32); let cl: i64 = rs_itoa(cb, n + 1)
149 let w: *i64 = ss_begin()
150 ss_add(w, 1, rk, val, vlen)
151 ss_add(w, 1, nk, cb, cl)
152 let rc: i64 = ss_commit(RS_PREFIX, w, rs_seg_next())
153 rs_unlock(lk)
154 if rc != 0 { return 0 - 1 }
155 return n
156}
157
158// SET (overwrite) row <ch>:<seq>=val in a locked commit (status mutation, e.g. PENDING->APPLIED).
159// last-wins: a new version of the same key shadows the old. Returns 0 ok, -1 commit-failed, -2 cap.
160func rs_row_set(ch: *u8, seq: i64, val: *u8, vlen: i64) -> i64 {
161 let lk: i64 = rs_lock()
162 if rs_segcount() >= RS_SEG_MAX {
163 rs_unlock(lk)
164 sys_write(2, "nx_role_store: REFUSED set -- roles store at segment cap; compaction rung needed\n" as *u8, 80)
165 return 0 - 2
166 }
167 let rk: *u8 = sys_mmap(128); rs_key(ch, seq, rk)
168 let w: *i64 = ss_begin()
169 ss_add(w, 1, rk, val, vlen)
170 let rc: i64 = ss_commit(RS_PREFIX, w, rs_seg_next())
171 rs_unlock(lk)
172 if rc != 0 { return 0 - 1 }
173 return 0
174}
175
176// "<role><suffix>" channel-name builders (suffix 'q'=queue, 'l'=ledger) into out (NUL-terminated).
177func rs_chan(role: *u8, suffix: i64, out: *u8) -> i64 {
178 var o: i64 = ss_cat(out, 0, role)
179 out[o] = suffix as u8; o = o + 1
180 out[o] = 0 as u8
181 return o
182}