code wiki / _hdl_build / nx_log_load.nx
nx_log_load.nx source
↩ module page · 248 lines · 8716 B
1// nx_log_load.nx -- IM7 (first slice): the team's append-only LOGS become
2// first-class records on the sovereign store (ONE substrate for corpus +
3// plan + capability evidence; the info-mgmt unification rung). Sources:
4// knowledge/status/pm_plan_durable.log (log=pm_plan)
5// knowledge/status/cap_registry_durable.log (log=capreg)
6// Each non-empty line -> canonical record {kind: team-log-line, log: <name>,
7// line: <text>} keyed by its OWN content CID (key = "log:" + nxc1-sha256 of
8// the canonical bytes):
9// idempotency BY CONSTRUCTION same line = same CID = same key+bytes =
10// dup, never re-added (law #10; re-run after
11// the log grows adds ONLY the new lines)
12// additive-only loader never deletes/overwrites
13// chunked commits 64 docs/segment keeps every segment far
14// under the (LOUD) postings caps; a loud cap
15// refusal aborts the run with the rc printed
16// Evidence: LOG-STORE lines -> stdout + knowledge/status/infomgmt_store.log,
17// plus term + AND probes proving the corpus is searchable. NO SQL.
18// license_tier: ORIGINAL
19
20import "nx_syscalls.nx"
21import "nx_canon_cid.nx"
22import "nx_seg_store.nx"
23const LG_MAGIC_4096: i64 = 4096
24const LG_MAGIC_8192: i64 = 8192
25const LG_MAGIC_1000000000: i64 = 1000000000
26const LG_MAGIC_4000: i64 = 4000
27const LG_MAGIC_4100: i64 = 4100
28
29func lg_p(s: *u8) -> i64 {
30 var n: i64 = 0
31 while s[n] != (0 as u8) { n = n + 1 }
32 sys_write(1, s, n)
33 return 0
34}
35
36func lg_fp(fd: i64, s: *u8) -> i64 {
37 var n: i64 = 0
38 while s[n] != (0 as u8) { n = n + 1 }
39 sys_write(fd, s, n)
40 return 0
41}
42
43func lg_fn(fd: i64, v: i64) -> i64 {
44 let bb: *u8 = sys_mmap(28)
45 var m: i64 = v
46 if m < 0 { m = 0 - m; sys_write(fd, "-" as *u8, 1) }
47 let t: *u8 = sys_mmap(28)
48 var k: i64 = 0
49 if m == 0 { t[0] = 48 as u8; k = 1 }
50 while m > 0 { t[k] = (48 + (m % 10)) as u8; m = m / 10; k = k + 1 }
51 var i: i64 = 0
52 while i < k { bb[i] = t[k - 1 - i]; i = i + 1 }
53 sys_write(fd, bb, k)
54 return 0
55}
56
57func lg_memeq(a: *u8, b: *u8, n: i64) -> i64 {
58 var i: i64 = 0
59 while i < n {
60 if a[i] != b[i] { return 0 }
61 i = i + 1
62 }
63 return 1
64}
65
66func lg_streq(a: *u8, b: *u8) -> i64 {
67 var i: i64 = 0
68 while 1 == 1 {
69 if a[i] != b[i] { return 0 }
70 if a[i] == (0 as u8) { return 1 }
71 i = i + 1
72 }
73 return 0
74}
75
76// run state shared across both source files (st layout):
77// st[0]=writer ptr st[1]=wdocs st[2]=segs_committed st[3]=added
78// st[4]=dup_infile st[5]=dup_instore st[6]=scanned st[7]=skipped_long
79// st[8]=seen pool ptr st[9]=nseen st[10]=handle ptr st[11]=abort rc
80const LG_CHUNK: i64 = 64
81const LG_SEENCAP: i64 = 4000
82
83func lg_flush(prefix: *u8, st: *i64) -> i64 {
84 if st[1] == 0 { return 0 }
85 let rc: i64 = ss_commit(prefix, st[0] as *i64, sys_now_us())
86 if rc != 0 { st[11] = rc; return rc }
87 st[0] = ss_begin() as i64
88 st[1] = 0
89 st[2] = st[2] + 1
90 return 0
91}
92
93func lg_load_file(prefix: *u8, path: *u8, logname: *u8, st: *i64) -> i64 {
94 let szp: *i64 = sys_mmap(16) as *i64
95 let b: *u8 = ss_readall(path, szp)
96 let sz: i64 = szp[0]
97 if sz <= 0 { return 0 - 1 }
98
99 let keys: *i64 = sys_mmap(8 * 4) as *i64
100 let vals: *i64 = sys_mmap(8 * 4) as *i64
101 keys[0] = "kind" as *u8 as i64
102 keys[1] = "log" as *u8 as i64
103 keys[2] = "line" as *u8 as i64
104 vals[0] = "team-log-line" as *u8 as i64
105 vals[1] = logname as i64
106
107 let lbuf: *u8 = sys_mmap(LG_MAGIC_4096)
108 let enc: *u8 = sys_mmap(LG_MAGIC_8192)
109 let cid: *u8 = sys_mmap(96)
110 let kbuf: *u8 = sys_mmap(128)
111 let pp: *i64 = sys_mmap(16) as *i64
112 let ll: *i64 = sys_mmap(16) as *i64
113 let h: *i64 = st[10] as *i64
114 let seen: *u8 = st[8] as *u8
115
116 var i: i64 = 0
117 while i < sz {
118 var e: i64 = i
119 while e < sz {
120 if b[e] == (10 as u8) { e = e + LG_MAGIC_1000000000 } else { e = e + 1 }
121 }
122 var lend: i64 = e
123 if e >= LG_MAGIC_1000000000 { lend = e - LG_MAGIC_1000000000 }
124 let llen: i64 = lend - i
125 if llen > LG_MAGIC_4000 {
126 st[7] = st[7] + 1
127 }
128 if llen > 0 { if llen <= LG_MAGIC_4000 {
129 st[6] = st[6] + 1
130 var t: i64 = 0
131 while t < llen { lbuf[t] = b[i + t]; t = t + 1 }
132 lbuf[llen] = 0 as u8
133 vals[2] = lbuf as i64
134 let el: i64 = canon_encode(keys, vals, 3, enc)
135 cid_of(enc, el, cid)
136 var ko: i64 = 0
137 ko = ss_cat(kbuf, ko, "log:" as *u8)
138 ko = ss_cat(kbuf, ko, cid)
139 kbuf[ko] = 0 as u8
140 // in-run dedup (covers segments committed during THIS run)
141 var dup: i64 = 0
142 var si: i64 = 0
143 while si < st[9] {
144 if dup == 0 {
145 if lg_streq((seen as i64 + si * 80) as *u8, kbuf) == 1 { dup = 1 }
146 }
147 si = si + 1
148 }
149 if dup == 1 { st[4] = st[4] + 1 }
150 if dup == 0 {
151 if st[9] >= LG_SEENCAP { st[11] = 0 - 99; return 0 - 2 }
152 var x: i64 = 0
153 let dst: *u8 = (seen as i64 + st[9] * 80) as *u8
154 while kbuf[x] != (0 as u8) { dst[x] = kbuf[x]; x = x + 1 }
155 dst[x] = 0 as u8
156 st[9] = st[9] + 1
157 // store dedup vs the run-start snapshot
158 var have: i64 = 0
159 let g: i64 = ss_hget(h, kbuf, pp, ll)
160 if g == 1 { if ll[0] == el { if lg_memeq(pp[0] as *u8, enc, el) == 1 { have = 1 } } }
161 if have == 1 { st[5] = st[5] + 1 }
162 if have == 0 {
163 if ss_add(st[0] as *i64, 1, kbuf, enc, el) != 0 { st[11] = 0 - 98; return 0 - 3 }
164 st[3] = st[3] + 1
165 st[1] = st[1] + 1
166 if st[1] >= LG_CHUNK {
167 if lg_flush(prefix, st) != 0 { return 0 - 4 }
168 }
169 }
170 }
171 } }
172 i = lend + 1
173 }
174 return 0
175}
176
177func main(argc: i64, argv: *i64) -> i64 {
178 var prefix: *u8 = "knowledge/store/im-" as *u8
179 if argc >= 2 { prefix = argv[1] as *u8 }
180 let logfd: i64 = sys_openat_append("knowledge/status/infomgmt_store.log" as *u8, 0x1a4)
181
182 let st: *i64 = sys_mmap(8 * 16) as *i64
183 st[0] = ss_begin() as i64
184 st[8] = sys_mmap(80 * LG_SEENCAP) as i64
185 st[10] = ss_open(prefix) as i64
186
187 let r1: i64 = lg_load_file(prefix, "knowledge/status/pm_plan_durable.log" as *u8, "pm_plan" as *u8, st)
188 if r1 != 0 {
189 lg_p("LOG-STORE pm_plan FAILED rc=" as *u8)
190 lg_fn(1, r1)
191 lg_p(" inner=" as *u8)
192 lg_fn(1, st[11])
193 lg_p(" -- fail loud\n" as *u8)
194 return 1
195 }
196 let r2: i64 = lg_load_file(prefix, "knowledge/status/cap_registry_durable.log" as *u8, "capreg" as *u8, st)
197 if r2 != 0 {
198 lg_p("LOG-STORE capreg FAILED rc=" as *u8)
199 lg_fn(1, r2)
200 lg_p(" inner=" as *u8)
201 lg_fn(1, st[11])
202 lg_p(" -- fail loud\n" as *u8)
203 return 1
204 }
205 if lg_flush(prefix, st) != 0 {
206 lg_p("LOG-STORE final commit FAILED -- fail loud\n" as *u8)
207 return 1
208 }
209
210 // searchable-corpus probes on a FRESH handle (sees this run's segments)
211 let h2: *i64 = ss_open(prefix)
212 let kp: *i64 = sys_mmap(8 * LG_MAGIC_4100) as *i64
213 let kl: *i64 = sys_mmap(8 * LG_MAGIC_4100) as *i64
214 let nterm: i64 = ss_term(h2, "infomgmt" as *u8, kp, kl, LG_MAGIC_4096)
215 let aterms: *i64 = sys_mmap(8 * 4) as *i64
216 aterms[0] = "infomgmt" as *u8 as i64
217 aterms[1] = "postings" as *u8 as i64
218 let nand: i64 = ss_term_and(h2, aterms, 2, kp, kl, LG_MAGIC_4096)
219
220 var fdi: i64 = 0
221 while fdi < 2 {
222 var fd: i64 = 1
223 if fdi == 1 { fd = logfd }
224 if fd > 0 {
225 lg_fp(fd, "LOG-STORE scanned=" as *u8)
226 lg_fn(fd, st[6])
227 lg_fp(fd, " added=" as *u8)
228 lg_fn(fd, st[3])
229 lg_fp(fd, " dup_infile=" as *u8)
230 lg_fn(fd, st[4])
231 lg_fp(fd, " dup_instore=" as *u8)
232 lg_fn(fd, st[5])
233 lg_fp(fd, " skipped_long=" as *u8)
234 lg_fn(fd, st[7])
235 lg_fp(fd, " segments=" as *u8)
236 lg_fn(fd, st[2])
237 lg_fp(fd, " term_infomgmt=" as *u8)
238 lg_fn(fd, nterm)
239 lg_fp(fd, " and_infomgmt_postings=" as *u8)
240 lg_fn(fd, nand)
241 lg_fp(fd, "\n" as *u8)
242 }
243 fdi = fdi + 1
244 }
245 if logfd > 0 { sys_close(logfd) }
246 if nterm <= 0 { return 1 }
247 return 0
248}