code wiki / _hdl_build / nx_cc_warc_ingest.nx
nx_cc_warc_ingest.nx source
↩ module page · 418 lines · 26892 B
1// nx_cc_warc_ingest.nx -- BULK Common Crawl ingester: a LOCAL .warc.gz file -> the dp-web-pub- web shard.
2// SCALE rung over nx_cc_ingest (one network round-trip per record): here ONE pre-downloaded WARC file
3// (~1GB, ~30-40k pages; each CC record = its own gzip member) is stream-walked member-by-member ->
4// 'response' records -> HTTP 200 + text/html -> html->text -> quality filters -> ss_add doc:/url:/loc:
5// rows + out: edges (same node identity as the crawler/serve, so every page feeds PageRank).
6// BATCH+RESUME BY CONSTRUCTION: the deflate path allocates interior mappings it cannot free, so one
7// process walks at most <maxmembers> members starting at the byte offset checkpointed in <localfile>.off,
8// commits its segments, rewrites the checkpoint, exits. A driver loops invocations until DONE prints.
9// (Also crash-safe: a killed batch never advances the checkpoint past committed work.)
10// usage: nx_cc_warc_ingest <localfile> <cc-warc-path> [maxmembers]
11// <cc-warc-path> = the crawl-data/... path the file came from; stored in loc: rows for the rot-fallback,
12// byte-offset-exact, so a served result can re-range-fetch its record just like nx_cc_ingest's rows.
13// exit 0 = batch clean (prints MORE nextoff=N or DONE, + verdict=PASS) ; nonzero = failure, checkpoint
14// NOT advanced past the bad byte (surface corruption, never silently skip). license_tier: ORIGINAL
15import "nx_corpus_ingest.nx" // ci_hash / ci_mkurlkey / dss_prefix / dss_mkkey / seg_store / nx_html_to_text
16import "nx_gzip_wrap.nx" // nx_gzip_inflate -> *NxGzipResult (bytes_consumed drives the member walk)
17import "nx_warc_reader.nx" // warc_next_record / wr_find_blankline / wr_header_val
18import "nx_outlink_harvest.nx" // olh_scan / olh_outkey -> P1 link-graph edges
19const CWI_MAGIC_1152: i64 = 1152
20const CWI_MAGIC_8192: i64 = 8192
21const CWI_MAGIC_1048576: i64 = 1048576
22const CWI_MAGIC_2048: i64 = 2048
23const CWI_MAGIC_2400: i64 = 2400
24
25// Batch geometry, MEASURED on real CC-MAIN-2026-25 data (2026-07-23): ~10KB avg compressed member
26// (request/metadata records are tiny), ~30 deflate interior mappings per member, 1000 members = 0.96s /
27// 186MB RSS. This WSL kernel's vm.max_map_count = 1,048,576 -> 12000 members (~360k maps) is comfortable;
28// the 128MiB window is what actually bounds a batch (~940MB WARC = ~8 batches).
29const CWI_WINDOW: i64 = 134217728 // read window per batch (128MiB)
30const CWI_MAXMEM: i64 = 12000 // default members per batch
31const CWI_RECCAP: i64 = 33554432 // largest single inflated WARC record accepted: a <=2MB compressed
32 // member can legitimately inflate 10-20MB (proven: member at abs
33 // 10005008 of CC-MAIN-...-00000 overflowed a 4MB cap); 32MB covers it
34const CWI_SEGCAP: i64 = 16777216 // 16MiB writer segments: serve cost grows with segment COUNT, so big
35 // segments keep the manifest short at scale (1MiB default -> ~150/warc)
36const CWI_MINTEXT: i64 = 400 // min extracted BYTES: kills nav-chrome stubs + soft-404s at ingest
37// MULTILINGUAL acceptance (2026-07-23, replaces the English-stopword filter that dropped ~55% of
38// responses): a page is real text if its first CWI_SCANSPAN bytes carry enough WORD MASS in any
39// supported script -- Latin (incl. folded accents), Cyrillic/Greek, CJK/kana/Hangul. CJK chars weigh
40// x2 (each carries ~a word-piece). Pages failing this are markup residue/symbol soup, not "some other
41// language" -- the tokenizer now searches all of the above natively (bigrams for CJK).
42const CWI_SCANSPAN: i64 = 4096 // classification window (bytes)
43const CWI_MINMASS: i64 = 150 // min weighted word-char count in the window
44const CWI_CJKW: i64 = 2 // CJK char weight in the mass sum
45const CWI_MAXURL: i64 = 1500 // == OLH_MAXURL policy: longer urls are tracker/junk, skip the page
46
47func cw_puts(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } sys_write(1, s, n); return 0 }
48func cw_num(v: i64) -> i64 {
49 let bb: *u8 = sys_mmap(28); var m: i64 = v
50 if m < 0 { sys_write(1, "-" as *u8, 1); m = 0 - m }
51 let t: *u8 = sys_mmap(28); var k: i64 = 0
52 if m == 0 { t[0] = 48 as u8; k = 1 }
53 while m > 0 { t[k] = (48 + (m % 10)) as u8; m = m / 10; k = k + 1 }
54 var i: i64 = 0; while i < k { bb[i] = t[k - 1 - i]; i = i + 1 }
55 sys_write(1, bb, k); return 0
56}
57func cw_atoin(s: *u8, n: i64) -> i64 {
58 var v: i64 = 0; var i: i64 = 0
59 while i < n { let c: i64 = s[i] as i64; if c >= 48 { if c <= 57 { v = v*10 + (c-48) } } i = i + 1 }
60 return v
61}
62// count occurrences of lit (litlen) in hay[0..n)
63func cw_cntsub(hay: *u8, n: i64, lit: *u8, litlen: i64) -> i64 {
64 var cnt: i64 = 0; var i: i64 = 0
65 while i + litlen <= n {
66 var j: i64 = 0; var hit: i64 = 1
67 while j < litlen { if hay[i+j] != lit[j] { hit = 0; j = litlen } else { j = j + 1 } }
68 if hit == 1 { cnt = cnt + 1; i = i + litlen } else { i = i + 1 }
69 }
70 return cnt
71}
72// literal substring present?
73func cw_hassub(hay: *u8, n: i64, lit: *u8, litlen: i64) -> i64 {
74 if cw_cntsub(hay, n, lit, litlen) > 0 { return 1 }
75 return 0
76}
77// script word-mass over text[0..n): counts[0]=latin counts[1]=cyrillic/greek counts[2]=cjk/kana/hangul.
78// Returns the weighted mass (lat + cyr + CWI_CJKW*cjk). Composes the tokenizer's OWN classifiers
79// (ss_u8cp/ss_fold_cp/ss_fold_word_cp/ss_is_cjk) so acceptance == searchability by construction.
80func cw_scriptmass(t: *u8, n: i64, counts: *i64) -> i64 {
81 counts[0] = 0; counts[1] = 0; counts[2] = 0
82 var i: i64 = 0
83 while i < n {
84 let c: i64 = t[i] as i64
85 if c < 128 {
86 if c >= 97 { if c <= 122 { counts[0] = counts[0] + 1 } }
87 if c >= 65 { if c <= 90 { counts[0] = counts[0] + 1 } }
88 i = i + 1
89 } else {
90 let v: i64 = ss_u8cp(t, n, i)
91 let cp: i64 = v >> 3
92 let adv: i64 = v & 7
93 if cp > 0 {
94 if ss_fold_cp(cp) != 0 { counts[0] = counts[0] + 1 } else {
95 if ss_fold_word_cp(cp) != 0 { counts[1] = counts[1] + 1 } else {
96 if ss_is_cjk(cp) == 1 { counts[2] = counts[2] + 1 }
97 }
98 }
99 }
100 i = i + adv
101 }
102 }
103 return counts[0] + counts[1] + CWI_CJKW * counts[2]
104}
105// "lng:<cid>" key builder (dominant-script tag row: lat|cyr|cjk -- the lang facet's data rung)
106func cw_lngkey(cid: i64, out: *u8) -> i64 {
107 out[0]=108 as u8; out[1]=110 as u8; out[2]=103 as u8; out[3]=58 as u8 // "lng:"
108 var o: i64=4; if cid==0 { out[o]=48 as u8; o=o+1; out[o]=0 as u8; return o }
109 let t: *u8=sys_mmap(24); var k: i64=0; var m: i64=cid
110 while m>0 { t[k]=(48+(m%10)) as u8; m=m/10; k=k+1 }
111 var j: i64=0; while j<k { out[o]=t[k-1-j]; o=o+1; j=j+1 } out[o]=0 as u8; return o
112}
113// "loc:<cid>" key builder (same shape as nx_cc_ingest's mk_lockey -- kept tiny+local because that organ
114// carries a main and cannot be imported)
115func cw_lockey(cid: i64, out: *u8) -> i64 {
116 out[0]=108 as u8; out[1]=111 as u8; out[2]=99 as u8; out[3]=58 as u8 // "loc:"
117 var o: i64=4; if cid==0 { out[o]=48 as u8; o=o+1; out[o]=0 as u8; return o }
118 let t: *u8=sys_mmap(24); var k: i64=0; var m: i64=cid
119 while m>0 { t[k]=(48+(m%10)) as u8; m=m/10; k=k+1 }
120 var j: i64=0; while j<k { out[o]=t[k-1-j]; o=o+1; j=j+1 } out[o]=0 as u8; return o
121}
122// read a small decimal state file; -1 if absent/unreadable
123func cw_readoff(path: *u8) -> i64 {
124 let fd: i64 = sys_openat_rd(path)
125 if fd < 0 { return 0 - 1 }
126 let b: *u8 = sys_mmap(32)
127 let n: i64 = sys_read(fd, b, 24)
128 sys_close(fd)
129 if n <= 0 { return 0 - 1 }
130 return cw_atoin(b, n)
131}
132// write a decimal to the state file (create/truncate)
133func cw_writeoff(path: *u8, v: i64) -> i64 {
134 let fd: i64 = sys_openat_wr(path, 420) // 0644
135 if fd < 0 { return 0 - 1 }
136 let t: *u8 = sys_mmap(28); var k: i64 = 0; var m: i64 = v
137 if m == 0 { t[0] = 48 as u8; k = 1 }
138 while m > 0 { t[k] = (48 + (m % 10)) as u8; m = m / 10; k = k + 1 }
139 let o: *u8 = sys_mmap(28); var i: i64 = 0
140 while i < k { o[i] = t[k-1-i]; i = i + 1 }
141 o[k] = 10 as u8
142 sys_write(fd, o, k+1)
143 sys_close(fd)
144 return 0
145}
146
147func main(argc: i64, argv: *i64) -> i64 {
148 if argc < 3 { cw_puts("usage: nx_cc_warc_ingest <localfile> <cc-warc-path> [maxmembers]\n" as *u8); return 1 }
149 let localfile: *u8 = argv[1] as *u8
150 let ccpath: *u8 = argv[2] as *u8
151 var maxmem: i64 = CWI_MAXMEM
152 if argc >= 4 { let a3: *u8 = argv[3] as *u8; var a3n: i64 = 0; while a3[a3n] != (0 as u8) { a3n = a3n + 1 } maxmem = cw_atoin(a3, a3n) }
153 if maxmem < 1 { maxmem = 1 }
154 if maxmem > CWI_MAXMEM { maxmem = CWI_MAXMEM }
155
156 // file size
157 let stb: *u8 = sys_mmap(160)
158 if sys_fstatat(localfile, stb) < 0 { cw_puts("WARC-INGEST cannot stat input\nverdict=FAIL\n" as *u8); return 2 }
159 let szp: *i64 = ((stb as i64) + 48) as *i64 // st_size @ +48 (x86_64 struct stat)
160 let filesize: i64 = szp[0]
161
162 // checkpoint path = localfile + ".off"
163 let offpath: *u8 = sys_mmap(CWI_MAGIC_1152)
164 var lp: i64 = 0; while localfile[lp] != (0 as u8) { offpath[lp] = localfile[lp]; lp = lp + 1 }
165 offpath[lp]=46 as u8; offpath[lp+1]=111 as u8; offpath[lp+2]=102 as u8; offpath[lp+3]=102 as u8; offpath[lp+4]=0 as u8
166 var startoff: i64 = cw_readoff(offpath)
167 if startoff < 0 { startoff = 0 }
168
169 cw_puts("=== nx_cc_warc_ingest r7-clamp " as *u8); cw_puts(ccpath); cw_puts(" size=" as *u8); cw_num(filesize)
170 cw_puts(" startoff=" as *u8); cw_num(startoff); cw_puts(" maxmembers=" as *u8); cw_num(maxmem); cw_puts(" ===\n" as *u8)
171 if startoff >= filesize {
172 cw_puts("WARC-INGEST DONE (already complete)\nverdict=PASS\n" as *u8); return 0
173 }
174
175 // read window
176 let fd: i64 = sys_openat_rd(localfile)
177 if fd < 0 { cw_puts("WARC-INGEST cannot open input\nverdict=FAIL\n" as *u8); return 2 }
178 sys_lseek(fd, startoff, 0)
179 let win: *u8 = sys_mmap(CWI_WINDOW)
180 var got: i64 = 0
181 var rdgo: i64 = 1
182 while rdgo == 1 {
183 let nr: i64 = sys_read(fd, (win as i64 + got) as *u8, CWI_WINDOW - got)
184 if nr <= 0 { rdgo = 0 } else { got = got + nr; if got >= CWI_WINDOW { rdgo = 0 } }
185 }
186 sys_close(fd)
187 var winfull: i64 = 0
188 if got == CWI_WINDOW { winfull = 1 }
189
190 // open web shard (dedup handle + fresh segid)
191 let prefix: *u8 = sys_mmap(512); dss_prefix("web" as *u8, prefix)
192 let segs: *i64 = sys_mmap(CWI_MAGIC_8192*8) as *i64
193 let nseg: i64 = ss_manifest_cap(prefix, segs, CWI_MAGIC_8192)
194 // R7 (2026-08-05): segid lives in a HEAP BOX, not a stack local. The clamp below is PROVEN
195 // present in the running binary (marker-build experiment: source changes track to the elf,
196 // and a byte-identical rebuild reproduces the deployed hash from clamped source) -- yet every
197 // deep-nested writer-full commit still received an ADDRESS-shaped segid (...540/...541,
198 // incrementing by one per commit, identical across runs = a deterministic mmap address).
199 // That is the known nx_cc deep-scope slot-aliasing class (the duplicate-let law, 1785933085):
200 // a stack local referenced 12 nesting levels down can silently read another slot. A heap box
201 // dereference cannot be re-slotted, so this sidesteps the compiler defect until the owner
202 // lane lands the add_local fix behind the equivalence net.
203 let sgb: *i64 = sys_mmap(16) as *i64
204 sgb[0] = 1
205 var si: i64 = 0
206 // ROOT FIX seq1730 (id 1785450987): segs[] holds POINTERS to seg-<id> name strings
207 // (nx_seg_store.nx:1234 stores segs[cnt] = name as i64), NOT ids -- so segs[si] + 1 produced
208 // MMAP_ADDRESS+1 (~1.4e14), the pointer-shaped poison that PINS a plane forever: every later
209 // epoch id sorts BELOW it in supersede order and its rows are silently shadowed while rc=0.
210 // Parse the DIGITS, as nx_web_shard_compact.nx:55-58 already does on this SAME array.
211 while si < nseg {
212 let sg_nm: *u8 = segs[si] as *u8
213 var sg_v: i64 = 0
214 var sg_ci: i64 = 0
215 while sg_nm[sg_ci] != (0 as u8) { let sg_c: i64 = sg_nm[sg_ci] as i64; if sg_c >= 48 { if sg_c <= 57 { sg_v = sg_v * 10 + (sg_c - 48) } } sg_ci = sg_ci + 1 }
216 // seq1730 REPAIR HALF (2026-08-05, mirrors ss_next_segid; mtime-touched 2026-08-05 PM --
217 // the builder reuses an artifact NEWER than the source, and the 08-05 AM fix edit landed
218 // AFTER that morning's build, so every later BUILT was a CACHE HIT of the pre-clamp
219 // binary; this touch forces the real compile): SKIP pointer-shaped ids in the
220 // max scan -- this manifest ALREADY holds address-named segments (seg-1407xx..., 0x7F mmap
221 // era), so max+1 was ADDRESS+1 and EVERY writer-full commit was REFUSED by the guard while
222 // the batch counters advanced (24 refusals across WARC 00010/00011, docs silently dropped).
223 if ss_segid_ok(sg_v) == 1 { if sg_v >= sgb[0] { sgb[0] = sg_v + 1 } }
224 si = si + 1
225 }
226 let h: *i64 = ss_open(prefix)
227 let pbox: *i64 = sys_mmap(16) as *i64; let lbox: *i64 = sys_mmap(16) as *i64
228 var w: *i64 = ss_begin_cap(CWI_SEGCAP)
229
230 // per-batch buffers (allocated ONCE; ss_add copies bytes so reuse is safe)
231 let text: *u8 = sys_mmap(CWI_MAGIC_1048576)
232 let urlv: *u8 = sys_mmap(CWI_MAGIC_2048)
233 let key: *u8 = sys_mmap(64); let ukey: *u8 = sys_mmap(64); let lkey: *u8 = sys_mmap(64); let okey: *u8 = sys_mmap(64)
234 let gkey: *u8 = sys_mmap(64)
235 let scounts: *i64 = sys_mmap(32) as *i64
236 let locv: *u8 = sys_mmap(CWI_MAGIC_2400)
237 let ceb: *i64 = sys_mmap(OLH_MAXEDGE * 8) as *i64
238 let fields: *i64 = sys_mmap(48) as *i64
239 let b2: *i64 = sys_mmap(32) as *i64
240
241 var off: i64 = 0 // window-relative walk position
242 var members: i64 = 0
243 var ingested: i64 = 0; var present: i64 = 0; var notresp: i64 = 0
244 var nothtml: i64 = 0; var thin: i64 = 0; var noneng: i64 = 0; var edged: i64 = 0
245 var committed: i64 = 0
246 var resyncs: i64 = 0 // undecodable members skipped via gzip-magic resync (loud, counted)
247 var badstop: i64 = 0 // 1 = corruption detected -> FAIL without advancing checkpoint
248 var go: i64 = 1
249 while go == 1 {
250 if members >= maxmem { go = 0 } else { if off >= got { go = 0 } else {
251 let gz: *NxGzipResult = nx_gzip_inflate((win as i64 + off) as *u8, got - off, CWI_RECCAP)
252 if gz.error_code != NX_GZ_OK {
253 // A member error is EITHER (a) the window edge cut it -- resume: stop the batch, the next
254 // batch re-reads this member with a FULL window -- or (b) a genuinely undecodable member.
255 // (b) is only PROVABLE when truncation is impossible: error at off==0 of a full window, or
256 // in an EOF window (no more file bytes exist). Then RESYNC: scan for the next gzip magic
257 // 1f 8b 08 and continue there -- loud, counted, never wedges. A mid-window error is ALWAYS
258 // treated as (a) first; if the member is truly bad the next batch hits it at off==0 and
259 // resyncs. This ordering never mistakes a truncated member's tail bytes for a member start.
260 if (gz.output_data as i64) != 0 { sys_munmap(gz.output_data, CWI_RECCAP) }
261 sys_munmap(gz as *u8, 96)
262 var provably_bad: i64 = 0
263 if off == 0 { provably_bad = 1 }
264 if winfull == 0 { provably_bad = 1 }
265 if provably_bad == 1 {
266 var ro: i64 = off + 3
267 var found: i64 = 0 - 1
268 while ro + 3 <= got {
269 if win[ro] == (31 as u8) { if win[ro+1] == (139 as u8) { if win[ro+2] == (8 as u8) { found = ro; ro = got } } }
270 if found < 0 { ro = ro + 1 }
271 }
272 if found >= 0 {
273 cw_puts(" RESYNC skipped undecodable member at abs " as *u8); cw_num(startoff + off)
274 cw_puts(" err=" as *u8); cw_num(gz.error_code)
275 cw_puts(" next-magic at abs " as *u8); cw_num(startoff + found); cw_puts("\n" as *u8)
276 resyncs = resyncs + 1
277 off = found
278 members = members + 1
279 } else {
280 if winfull == 1 { badstop = 1; go = 0 } else {
281 // EOF window, no further member start: clean end if only trailing scraps remain
282 if got - off < 64 { off = got; go = 0 } else { badstop = 1; go = 0 }
283 }
284 }
285 } else {
286 go = 0
287 }
288 } else {
289 let mem: *u8 = gz.output_data
290 let memn: i64 = gz.output_size
291 let msize: i64 = gz.bytes_consumed
292 // ---- process one WARC record ----
293 let nr2: i64 = warc_next_record(mem, memn, 0, fields)
294 if nr2 >= 0 { if fields[1] == 8 {
295 // WARC-Type value == "response"?
296 let tp: i64 = fields[0]
297 var isresp: i64 = 0
298 if mem[tp]==(114 as u8) { if mem[tp+1]==(101 as u8) { if mem[tp+2]==(115 as u8) { if mem[tp+3]==(112 as u8) { isresp = 1 } } } }
299 if isresp == 1 {
300 let ul: i64 = fields[3]
301 if ul > 0 { if ul < CWI_MAXURL {
302 var ui: i64 = 0; let uo: i64 = fields[2]
303 while ui < ul { urlv[ui] = mem[uo+ui]; ui = ui + 1 } urlv[ul] = 0 as u8
304 var skipu: i64 = 0
305 if cw_hassub(urlv, ul, "robots.txt" as *u8, 10) == 1 { skipu = 1 }
306 if ul >= 4 { if urlv[ul-4]==(46 as u8) { if urlv[ul-3]==(120 as u8) { if urlv[ul-2]==(109 as u8) { if urlv[ul-1]==(108 as u8) { skipu = 1 } } } } }
307 let coff: i64 = fields[4]
308 var clen: i64 = fields[5]
309 if coff + clen > memn { clen = memn - coff }
310 if skipu == 0 { if clen > 32 {
311 // HTTP status: "HTTP/1.x NNN" -> bytes 9..11 of the content block
312 var is200: i64 = 0
313 if mem[coff]==(72 as u8) { if mem[coff+9]==(50 as u8) { if mem[coff+10]==(48 as u8) { if mem[coff+11]==(48 as u8) { is200 = 1 } } } }
314 if is200 == 1 {
315 let hb: i64 = wr_find_blankline(mem, coff, coff + clen)
316 if hb > 0 {
317 var ishtml: i64 = 0
318 if wr_header_val(mem, coff, hb + 2, memn, "content-type" as *u8, 12, b2) == 1 {
319 let cv: i64 = b2[0]
320 if mem[cv]==(116 as u8) { if mem[cv+1]==(101 as u8) { if mem[cv+2]==(120 as u8) { if mem[cv+3]==(116 as u8) { if mem[cv+4]==(47 as u8) { if mem[cv+5]==(104 as u8) { ishtml = 1 } } } } } }
321 }
322 if ishtml == 0 { nothtml = nothtml + 1 } else {
323 let body: *u8 = (mem as i64 + hb + 4) as *u8
324 let blen: i64 = coff + clen - (hb + 4)
325 let tlen: i64 = nx_html_to_text(body, blen, text, CWI_MAGIC_1048576)
326 if tlen < CWI_MINTEXT { thin = thin + 1 } else {
327 var espan: i64 = tlen; if espan > CWI_SCANSPAN { espan = CWI_SCANSPAN }
328 let smass: i64 = cw_scriptmass(text, espan, scounts)
329 if smass < CWI_MINMASS { noneng = noneng + 1 } else {
330 var tn: i64 = tlen; if tn > CI_DOCCAP { tn = CI_DOCCAP }
331 let cid: i64 = ci_hash(text, tn)
332 dss_mkkey(cid, key)
333 var already: i64 = 0
334 if (h as i64) != 0 { if ss_hget(h, key, pbox, lbox) == 1 { already = 1 } }
335 if already == 1 { present = present + 1 } else {
336 if ss_add(w, 1, key, text, tn) < 0 { if ss_commit(prefix, w, sgb[0])==0 { committed=committed+1 } sgb[0]=sgb[0]+1; w[1]=0; ss_add(w, 1, key, text, tn) }
337 ci_mkurlkey(cid, ukey)
338 if ss_add(w, 1, ukey, urlv, ul) < 0 { if ss_commit(prefix, w, sgb[0])==0 { committed=committed+1 } sgb[0]=sgb[0]+1; w[1]=0; ss_add(w, 1, ukey, urlv, ul) }
339 // loc: "<ccpath>|<abs-member-off>|<member-len>" = byte-exact rot-fallback locator
340 var lo: i64 = 0
341 var lf: i64 = 0; while ccpath[lf] != (0 as u8) { locv[lo]=ccpath[lf]; lo=lo+1; lf=lf+1 }
342 locv[lo]=124 as u8; lo=lo+1
343 let absoff: i64 = startoff + off
344 let t2: *u8=sys_mmap(24); var k2: i64=0; var m2: i64=absoff
345 if m2==0 { t2[0]=48 as u8; k2=1 }
346 while m2>0 { t2[k2]=(48+(m2%10)) as u8; m2=m2/10; k2=k2+1 }
347 var j2: i64=0; while j2<k2 { locv[lo]=t2[k2-1-j2]; lo=lo+1; j2=j2+1 }
348 locv[lo]=124 as u8; lo=lo+1
349 var k3: i64=0; var m3: i64=msize
350 if m3==0 { t2[0]=48 as u8; k3=1 }
351 while m3>0 { t2[k3]=(48+(m3%10)) as u8; m3=m3/10; k3=k3+1 }
352 var j3: i64=0; while j3<k3 { locv[lo]=t2[k3-1-j3]; lo=lo+1; j3=j3+1 }
353 cw_lockey(cid, lkey)
354 if ss_add(w, 1, lkey, locv, lo) < 0 { if ss_commit(prefix, w, sgb[0])==0 { committed=committed+1 } sgb[0]=sgb[0]+1; w[1]=0; ss_add(w, 1, lkey, locv, lo) }
355 // dominant-script tag (lat|cyr|cjk): the lang-facet data rung
356 var tagp: *u8 = "lat" as *u8
357 if scounts[1] > scounts[0] { if scounts[1] >= CWI_CJKW * scounts[2] { tagp = "cyr" as *u8 } }
358 if CWI_CJKW * scounts[2] > scounts[0] { if CWI_CJKW * scounts[2] > scounts[1] { tagp = "cjk" as *u8 } }
359 cw_lngkey(cid, gkey)
360 if ss_add(w, 1, gkey, tagp, 3) < 0 { if ss_commit(prefix, w, sgb[0])==0 { committed=committed+1 } sgb[0]=sgb[0]+1; w[1]=0; ss_add(w, 1, gkey, tagp, 3) }
361 // link-graph edges: node identity = ci_hash(url) == dss_urlcid at serve
362 let cne: i64 = olh_scan(body, blen, urlv, ul, ceb, OLH_MAXEDGE)
363 if cne > 0 {
364 olh_outkey(ci_hash(urlv, ul), okey)
365 if ss_add(w, 1, okey, ceb as *u8, cne * 8) < 0 { if ss_commit(prefix, w, sgb[0])==0 { committed=committed+1 } sgb[0]=sgb[0]+1; w[1]=0; ss_add(w, 1, okey, ceb as *u8, cne * 8) }
366 edged = edged + 1
367 }
368 ingested = ingested + 1
369 if (ingested % 200) == 0 { cw_puts(" ... ingested=" as *u8); cw_num(ingested); cw_puts(" members=" as *u8); cw_num(members); cw_puts("\n" as *u8) }
370 }
371 }
372 }
373 }
374 }
375 }
376 } }
377 } }
378 } else { notresp = notresp + 1 }
379 } }
380 // ---- free the member's big buffers, advance ----
381 sys_munmap(gz.output_data, CWI_RECCAP)
382 sys_munmap(gz as *u8, 96)
383 off = off + msize
384 members = members + 1
385 }
386 } }
387 }
388 if w[1] > 0 { if ss_commit(prefix, w, sgb[0])==0 { committed = committed + 1 } }
389
390 if badstop == 1 {
391 cw_puts("WARC-INGEST corrupt/truncated member at abs offset " as *u8); cw_num(startoff + off)
392 cw_puts(" (checkpoint NOT advanced past committed work)\n" as *u8)
393 // still checkpoint what we cleanly finished so a re-run resumes here, not at 0
394 cw_writeoff(offpath, startoff + off)
395 cw_puts("verdict=FAIL\n" as *u8)
396 return 3
397 }
398 let newoff: i64 = startoff + off
399 cw_writeoff(offpath, newoff)
400 cw_puts("WARC-INGEST batch: members=" as *u8); cw_num(members)
401 cw_puts(" ingested=" as *u8); cw_num(ingested)
402 cw_puts(" present=" as *u8); cw_num(present)
403 cw_puts(" edged=" as *u8); cw_num(edged)
404 cw_puts(" notresp=" as *u8); cw_num(notresp)
405 cw_puts(" nothtml=" as *u8); cw_num(nothtml)
406 cw_puts(" thin=" as *u8); cw_num(thin)
407 cw_puts(" lowmass=" as *u8); cw_num(noneng)
408 cw_puts(" segs=" as *u8); cw_num(committed)
409 cw_puts(" resyncs=" as *u8); cw_num(resyncs)
410 cw_puts("\n" as *u8)
411 if newoff >= filesize {
412 cw_puts("WARC-INGEST DONE total-file-consumed\n" as *u8)
413 } else {
414 cw_puts("WARC-INGEST MORE nextoff=" as *u8); cw_num(newoff); cw_puts("\n" as *u8)
415 }
416 cw_puts("verdict=PASS\n" as *u8)
417 return 0
418}