nx_fp_backfill.nx source

1// nx_fp_backfill.nx -- BACKFILL THE fp: FINGERPRINT ROW OVER ALREADY-INGESTED DOCS. 2// 3// WHY: nx_web_crawl_step now persists fp:<cid> (the simhash it always computed and used to discard), 4// so intra-host near-duplication -- the real doorway signature -- is a property of the index. But that 5// is FORWARD-ONLY: every doc ingested before 2026-08-16 has no fp: row, so a duplication census would 6// silently measure only the newest slice and report it as the corpus. This closes that gap. 7// 8// NO NETWORK. The text is already stored; this reads doc:<cid>, hashes it, writes fp:<cid>. A source 9// that died years ago costs nothing here -- which is the whole argument for an entity corpus over a 10// page index: capture once, and every later question is local. 11// 12// MEMORY SHAPE, STATED because the sibling's is different: nx_web_shard_compact was OOM-killed twice 13// (bytes-in=1883885124, 2026-07-25) because it MATERIALISES a merged segment in RAM. This does not -- 14// it holds one doc at a time and emits a ~20-byte row, so its bound is on RUNTIME, not memory. 15// 16// BOUNDED AND RESUMABLE, never a silent runaway: 17// argv[1] max REQUIRED and must be > 0 -- refuse rather than default to unbounded 18// argv[2] skip resume offset into the walk order (default 0) 19// argv[3] commit literal "commit" to write; DEFAULT IS A DRY RUN that changes nothing 20// PROGRESS IS PROVABLE: every run prints scanned/had_fp/written/capped. The sibling's range-fold note 21// is the warning -- a bounded sweep with the wrong selection rule runs forever WITHOUT ADVANCING, so 22// "did this converge" must be a number, not a feeling. capped=1 says the run stopped at max, i.e. the 23// corpus is NOT yet fully covered and the next run needs skip=<scanned+skip>. 24// license_tier: ORIGINAL 25import "nx_corpus_ingest.nx" // dss_prefix / dss_mkkey / ci_mkfpkey / ss_* primitives 26import "nx_simhash.nx" // nx_simhash_fingerprint -- the SAME hasher the crawler uses, composed 27import "nx_itoa_lib.nx" // ccz_cat_num -- THE canonical integer emitter 28 29const FB_KEYCAP: i64 = 64 30const FB_VALCAP: i64 = 32 31 32func fb_puts(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } sys_write(1, s, n); return 0 } 33func fb_num(v: i64) -> i64 { 34 if v == 0 { fb_puts("0" as *u8); return 0 } 35 var m: i64 = v 36 if m < 0 { fb_puts("-" as *u8); m = 0 - m } 37 let d: *u8 = sys_mmap(24); var k: i64 = 0 38 while m > 0 { d[k] = (48 + (m % 10)) as u8; m = m / 10; k = k + 1 } 39 let o: *u8 = sys_mmap(24); var i: i64 = 0 40 while i < k { o[i] = d[k - 1 - i]; i = i + 1 } 41 sys_write(1, o, k) 42 return 0 43} 44func fb_atoi(s: *u8) -> i64 { 45 var v: i64 = 0 46 var i: i64 = 0 47 while s[i] != (0 as u8) { let c: i64 = s[i] as i64; if c >= 48 { if c <= 57 { v = v * 10 + (c - 48) } } i = i + 1 } 48 return v 49} 50 51func main(argc: i64, argv: *i64) -> i64 { 52 if argc < 2 { fb_puts("usage: nx_fp_backfill <max> [skip] [commit] (max REQUIRED > 0; default is a DRY RUN)\n" as *u8); return 2 } 53 let maxn: i64 = fb_atoi(argv[1] as *u8) 54 if maxn <= 0 { fb_puts("REFUSED: max must be > 0 -- an unbounded sweep over this shard is how the sibling got OOM-killed\n" as *u8); return 2 } 55 var skip: i64 = 0 56 if argc >= 3 { skip = fb_atoi(argv[2] as *u8) } 57 var docommit: i64 = 0 58 if argc >= 4 { let a3: *u8 = argv[3] as *u8; if a3[0] == (99 as u8) { docommit = 1 } } 59 60 let prefix: *u8 = sys_mmap(512) 61 dss_prefix("web" as *u8, prefix) 62 let h: *i64 = ss_open(prefix) 63 if (h as i64) == 0 { fb_puts("BACKFILL FAIL: web shard absent\n" as *u8); return 3 } 64 let w: *i64 = ss_begin() 65 let segsbox: *i64 = sys_mmap(16) as *i64 66 segsbox[0] = 0 67 let segidbox: *i64 = sys_mmap(16) as *i64 68 segidbox[0] = ss_next_segid(prefix) 69 70 let kbuf: *u8 = sys_mmap(FB_KEYCAP) 71 let dkey: *u8 = sys_mmap(FB_KEYCAP) 72 let fkey: *u8 = sys_mmap(FB_KEYCAP) 73 let fval: *u8 = sys_mmap(FB_VALCAP) 74 let pp: *i64 = sys_mmap(16) as *i64 75 let pl: *i64 = sys_mmap(16) as *i64 76 let dp: *i64 = sys_mmap(16) as *i64 77 let dl: *i64 = sys_mmap(16) as *i64 78 79 var seen: i64 = 0 // url: rows encountered (walk position, for the resume offset) 80 var scanned: i64 = 0 // rows actually examined this run 81 var had: i64 = 0 // already carried an fp: row 82 var wrote: i64 = 0 // fp: rows emitted 83 var nodoc: i64 = 0 // url: row with no readable doc: row -- reported, never silently skipped 84 var capped: i64 = 0 85 86 let ns: i64 = h[0] 87 var s: i64 = 0 88 while s < ns { 89 let kb: *u8 = h[1 + 8 * s] as *u8 90 if h[2 + 8 * s] >= 8 { 91 let m9: i64 = ss_r32(kb, 4) 92 var e9: i64 = 0 93 while e9 < m9 { 94 let eo: i64 = 8 + 4 * m9 + ss_r32(kb, 8 + 4 * e9) 95 if (kb[eo] as i64) == 1 { 96 let kl9: i64 = ss_r32(kb, eo + 1) 97 if kl9 >= 5 { if kl9 < 60 { 98 if kb[eo + 5] == (117 as u8) { if kb[eo + 6] == (114 as u8) { if kb[eo + 7] == (108 as u8) { if kb[eo + 8] == (58 as u8) { 99 seen = seen + 1 100 if seen > skip { if capped == 0 { 101 if scanned >= maxn { capped = 1 } else { 102 scanned = scanned + 1 103 var c: i64 = 0 104 while c < kl9 { kbuf[c] = kb[eo + 5 + c]; c = c + 1 } 105 kbuf[kl9] = 0 as u8 106 var cid: i64 = 0 107 var ki: i64 = 4 108 while ki < kl9 { let kc: i64 = kbuf[ki] as i64; if kc >= 48 { if kc <= 57 { cid = cid * 10 + (kc - 48) } } ki = ki + 1 } 109 ci_mkfpkey(cid, fkey) 110 var present: i64 = 0 111 if ss_hget(h, fkey, pp, pl) == 1 { if pl[0] > 0 { present = 1 } } 112 if present == 1 { had = had + 1 } else { 113 dss_mkkey(cid, dkey) 114 var got: i64 = 0 115 if ss_hget(h, dkey, dp, dl) == 1 { if dl[0] > 0 { got = 1 } } 116 if got == 0 { nodoc = nodoc + 1 } else { 117 let fp: i64 = nx_simhash_fingerprint(dp[0] as *u8, dl[0]) 118 let fn2: i64 = ccz_cat_num(fval, 0, fp) 119 wrote = wrote + 1 120 if docommit == 1 { 121 if ss_add(w, 1, fkey, fval, fn2) < 0 { 122 if ss_commit(prefix, w, segidbox[0]) == 0 { segsbox[0] = segsbox[0] + 1 } 123 segidbox[0] = segidbox[0] + 1 124 w[1] = 0 125 ss_add(w, 1, fkey, fval, fn2) 126 } 127 } 128 } 129 } 130 } 131 } } 132 } } } } 133 } } 134 } 135 e9 = e9 + 1 136 } 137 } 138 s = s + 1 139 } 140 if docommit == 1 { if ss_commit(prefix, w, segidbox[0]) == 0 { segsbox[0] = segsbox[0] + 1 } } 141 142 fb_puts("{\"tool\":\"nx_fp_backfill\",\"url_rows_walked\":" as *u8); fb_num(seen) 143 fb_puts(",\"skip\":" as *u8); fb_num(skip) 144 fb_puts(",\"scanned\":" as *u8); fb_num(scanned) 145 fb_puts(",\"had_fp\":" as *u8); fb_num(had) 146 fb_puts(",\"wrote_fp\":" as *u8); fb_num(wrote) 147 fb_puts(",\"no_doc_row\":" as *u8); fb_num(nodoc) 148 fb_puts(",\"segments_committed\":" as *u8); fb_num(segsbox[0]) 149 fb_puts(",\"committed\":" as *u8); fb_num(docommit) 150 fb_puts(",\"capped\":" as *u8); fb_num(capped) 151 fb_puts("}\n" as *u8) 152 // The partition MUST reconcile: every scanned row is exactly one of had/wrote/no_doc. 153 if had + wrote + nodoc == scanned { fb_puts("PARTITION-RECONCILES\n" as *u8) } else { fb_puts("PARTITION-LEAK (had+wrote+no_doc != scanned) -- do not trust these counts\n" as *u8) } 154 if capped == 1 { fb_puts("CAPPED at max -- corpus NOT fully covered; next run: skip=" as *u8); fb_num(skip + scanned); fb_puts("\n" as *u8) } 155 fb_puts("FP-BACKFILL-OK\n" as *u8) 156 return 0 157}