code wiki / _hdl_build / nx_census_load.nx
nx_census_load.nx source
↩ module page · 228 lines · 7931 B
1// nx_census_load.nx -- IM5: the FIRST REAL CORPUS on the sovereign segment
2// store. Parses knowledge/status/elder_census.log (CENSUS-SVC name=<n>
3// class=<c> lines -- 447 lines = 149 services x 3 census runs, REAL duplicate
4// data) into canonical CID records on the DURABLE store knowledge/store/im-*.
5//
6// Laws exercised on real data:
7// dedup-by-CID a record whose canonical bytes already live under its key
8// is SKIPPED (the 3x census repeats collapse measurably)
9// idempotency run twice => second run adds 0 records, commits 0 segments
10// (safe-to-run-twice law #10)
11// additive-only loader never deletes/overwrites; new states would land as
12// new versions in new segments
13// Evidence: CENSUS-STORE lines -> stdout + knowledge/status/infomgmt_store.log,
14// plus an indexed-read probe of a known service. Pure NishiLang, NO SQL.
15// license_tier: ORIGINAL
16
17import "nx_syscalls.nx"
18import "nx_canon_cid.nx"
19import "nx_seg_store.nx"
20const K_MAGIC_1024: i64 = 1024
21const K_MAGIC_1000000000: i64 = 1000000000
22const K_MAGIC_2048: i64 = 2048
23
24func cl_p(s: *u8) -> i64 {
25 var n: i64 = 0
26 while s[n] != (0 as u8) { n = n + 1 }
27 sys_write(1, s, n)
28 return 0
29}
30
31func cl_fp(fd: i64, s: *u8) -> i64 {
32 var n: i64 = 0
33 while s[n] != (0 as u8) { n = n + 1 }
34 sys_write(fd, s, n)
35 return 0
36}
37
38func cl_fn(fd: i64, v: i64) -> i64 {
39 let bb: *u8 = sys_mmap(28)
40 var m: i64 = v
41 if m < 0 { m = 0 - m; sys_write(fd, "-" as *u8, 1) }
42 let t: *u8 = sys_mmap(28)
43 var k: i64 = 0
44 if m == 0 { t[0] = 48 as u8; k = 1 }
45 while m > 0 { t[k] = (48 + (m % 10)) as u8; m = m / 10; k = k + 1 }
46 var i: i64 = 0
47 while i < k { bb[i] = t[k - 1 - i]; i = i + 1 }
48 sys_write(fd, bb, k)
49 return 0
50}
51
52// does line at b[off..] start with the literal s? returns matched length or -1
53func cl_starts(b: *u8, off: i64, lim: i64, s: *u8) -> i64 {
54 var i: i64 = 0
55 while s[i] != (0 as u8) {
56 if off + i >= lim { return 0 - 1 }
57 if b[off + i] != s[i] { return 0 - 1 }
58 i = i + 1
59 }
60 return i
61}
62
63func cl_memeq(a: *u8, b: *u8, n: i64) -> i64 {
64 var i: i64 = 0
65 while i < n {
66 if a[i] != b[i] { return 0 }
67 i = i + 1
68 }
69 return 1
70}
71
72func main() -> i64 {
73 let prefix: *u8 = "knowledge/store/im-" as *u8
74 let logfd: i64 = sys_openat_append("knowledge/status/infomgmt_store.log" as *u8, 0x1a4)
75
76 let szp: *i64 = sys_mmap(16) as *i64
77 let b: *u8 = ss_readall("knowledge/status/elder_census.log" as *u8, szp)
78 let sz: i64 = szp[0]
79 if sz <= 0 {
80 cl_p("CENSUS-STORE source=elder_census.log MISSING -- fail loud\n" as *u8)
81 return 1
82 }
83
84 let w: *i64 = ss_begin()
85 // per-run seen-key list (in-file 3x repeats dedup before the store is asked)
86 let seen: *i64 = sys_mmap(8 * K_MAGIC_1024) as *i64
87 var nseen: i64 = 0
88 var scanned: i64 = 0
89 var added: i64 = 0
90 var dup_infile: i64 = 0
91 var dup_instore: i64 = 0
92
93 let keys: *i64 = sys_mmap(8 * 4) as *i64
94 let vals: *i64 = sys_mmap(8 * 4) as *i64
95 keys[0] = "kind" as *u8 as i64
96 keys[1] = "name" as *u8 as i64
97 keys[2] = "class" as *u8 as i64
98 vals[0] = "elder-service" as *u8 as i64
99
100 let pp: *i64 = sys_mmap(16) as *i64
101 let ll: *i64 = sys_mmap(16) as *i64
102
103 var i: i64 = 0
104 while i < sz {
105 // line start at i; find end
106 var e: i64 = i
107 while e < sz {
108 if b[e] == (10 as u8) { e = e + K_MAGIC_1000000000 } else { e = e + 1 }
109 }
110 var lend: i64 = e
111 if e >= K_MAGIC_1000000000 { lend = e - K_MAGIC_1000000000 }
112 let m: i64 = cl_starts(b, i, lend, "CENSUS-SVC name=" as *u8)
113 if m > 0 {
114 scanned = scanned + 1
115 // name until space; p ends at the char AFTER the space
116 var p: i64 = i + m
117 let name: *u8 = sys_mmap(160)
118 var t: i64 = 0
119 var go2: i64 = 1
120 while go2 == 1 {
121 if p >= lend { go2 = 0 }
122 if go2 == 1 {
123 if b[p] == (32 as u8) { p = p + 1; go2 = 0 } else {
124 if t < 150 { name[t] = b[p]; t = t + 1 }
125 p = p + 1
126 }
127 }
128 }
129 name[t] = 0 as u8
130 // class= after the space
131 let m2: i64 = cl_starts(b, p, lend, "class=" as *u8)
132 let cls: *u8 = sys_mmap(64)
133 var c: i64 = 0
134 if m2 > 0 {
135 var p2: i64 = p + m2
136 while p2 < lend {
137 if c < 60 { cls[c] = b[p2]; c = c + 1 }
138 p2 = p2 + 1
139 }
140 }
141 cls[c] = 0 as u8
142 if t > 0 { if c > 0 {
143 // in-run dedup
144 var dup: i64 = 0
145 var si: i64 = 0
146 while si < nseen {
147 if dup == 0 {
148 var eq2: i64 = 1
149 let sp: *u8 = seen[si] as *u8
150 var x: i64 = 0
151 while eq2 == 1 {
152 if sp[x] != name[x] { eq2 = 0 }
153 if eq2 == 1 { if sp[x] == (0 as u8) { dup = 1; eq2 = 0 } }
154 x = x + 1
155 }
156 }
157 si = si + 1
158 }
159 if dup == 1 { dup_infile = dup_infile + 1 }
160 if dup == 0 {
161 if nseen < 1000 { seen[nseen] = name as i64; nseen = nseen + 1 }
162 // canonical record + key
163 vals[1] = name as i64
164 vals[2] = cls as i64
165 let enc: *u8 = sys_mmap(K_MAGIC_2048)
166 let el: i64 = canon_encode(keys, vals, 3, enc)
167 let kbuf: *u8 = sys_mmap(192)
168 var ko: i64 = 0
169 ko = ss_cat(kbuf, ko, "svc:" as *u8)
170 ko = ss_cat(kbuf, ko, name)
171 kbuf[ko] = 0 as u8
172 // dedup-by-CID vs the store (canonical bytes equal = same CID)
173 var have: i64 = 0
174 let g: i64 = ss_get_idx(prefix, kbuf, pp, ll)
175 if g == 1 { if ll[0] == el { if cl_memeq(pp[0] as *u8, enc, el) == 1 { have = 1 } } }
176 if have == 1 { dup_instore = dup_instore + 1 }
177 if have == 0 {
178 ss_add(w, 1, kbuf, enc, el)
179 added = added + 1
180 }
181 }
182 } }
183 }
184 i = lend + 1
185 }
186
187 var committed: i64 = 0
188 if added > 0 {
189 let rc: i64 = ss_commit(prefix, w, sys_now_us())
190 if rc != 0 {
191 cl_p("CENSUS-STORE commit FAILED rc-loud\n" as *u8)
192 return 1
193 }
194 committed = 1
195 }
196
197 // indexed-read probe of a known service (evidence the corpus is queryable)
198 var probe: i64 = 0
199 let g2: i64 = ss_get_idx(prefix, "svc:svc-wardrobe" as *u8, pp, ll)
200 if g2 == 1 { probe = 1 }
201
202 var fdi: i64 = 0
203 while fdi < 2 {
204 var fd: i64 = 1
205 if fdi == 1 { fd = logfd }
206 if fd > 0 {
207 cl_fp(fd, "CENSUS-STORE epoch=" as *u8)
208 cl_fn(fd, sys_now_realtime_sec())
209 cl_fp(fd, " scanned=" as *u8)
210 cl_fn(fd, scanned)
211 cl_fp(fd, " unique_new=" as *u8)
212 cl_fn(fd, added)
213 cl_fp(fd, " dup_infile=" as *u8)
214 cl_fn(fd, dup_infile)
215 cl_fp(fd, " dup_instore=" as *u8)
216 cl_fn(fd, dup_instore)
217 cl_fp(fd, " segments_added=" as *u8)
218 cl_fn(fd, committed)
219 cl_fp(fd, " probe_wardrobe=" as *u8)
220 cl_fn(fd, probe)
221 cl_fp(fd, "\n" as *u8)
222 }
223 fdi = fdi + 1
224 }
225 if logfd > 0 { sys_close(logfd) }
226 if probe == 1 { return 0 }
227 return 1
228}