nx_research_fetch_ext.nx source
↩ module page · 97 lines · 5536 B
1// nx_research_fetch_ext.nx -- DYNAMIC / GROWABLE fetch extensions (operator 2026-07-07: "make the fetcher
2// dynamic and we can keep growing and keep going"). Additive lib over nx_research_engine (no main).
3// rf_fetch_bank_paged: fetch an offset/length-paginated API (HF datasets-server) across N pages, appending
4// each raw HTTP response to knowledge/fetched/<name>.raw. The record-scanning benchmark parsers
5// (db_find_key "question"/"context") skip the intervening HTTP headers and find records across ALL pages,
6// so concatenation "just works" -> the researcher's eval sets grow to any size with bounded memory.
7// license_tier: ORIGINAL
8import "nx_research_engine.nx"
9const K_MAGIC_4096: i64 = 4096
10const K_MAGIC_8388608: i64 = 8388608
11
12// append literal s to dst[off..]; return new off
13func rfx_cat(dst: *u8, off: i64, s: *u8) -> i64 {
14 var i: i64 = 0
15 while s[i] != (0 as u8) { dst[off+i] = s[i]; i = i + 1 }
16 return off + i
17}
18// write unsigned int as decimal into dst[off..]; return new off
19func rfx_uitoa(dst: *u8, off: i64, v: i64) -> i64 {
20 if v == 0 { dst[off] = (48 as u8); return off + 1 }
21 let tmp: *u8 = sys_mmap(24)
22 var m: i64 = v
23 var k: i64 = 0
24 while m > 0 { tmp[k] = (48 + (m%10)) as u8; m = m/10; k = k + 1 }
25 var i: i64 = 0
26 while i < k { dst[off+i] = tmp[k-1-i]; i = i + 1 }
27 return off + k
28}
29
30// fetch urlbase + "&offset=<p*pagesize>&length=<pagesize>" for p in 0..npages; concat page BODIES -> <name>.raw.
31// urlbase must NOT already carry offset/length. HARDENED (debt eaten 2026-07-07):
32// IDEMPOTENT -- <name>.raw already banked -> [banked-skip] (matches rf_fetch_bank).
33// ATOMIC -- writes <name>.raw.part, sys_renameat -> .raw ONLY when ALL pages banked (the ecosystem
34// tmp+rename pattern). A failed run leaves .part (inspectable), NEVER a partial .raw that a
35// harness would silently score short.
36// ALL-OR-FAIL -- returns 1 only when okpages==npages; any failed page -> 0.
37// TRUNC TOOTH -- a page that FILLS the buffer (n>=cap) is treated as truncated -> failed, not banked.
38func rf_fetch_bank_paged(urlbase: *u8, name: *u8, store: *TrustStore, pagesize: i64, npages: i64) -> i64 {
39 return rf_fetch_bank_paged_from(urlbase, name, store, pagesize, npages, 0)
40}
41
42// startpage variant (ADDITIVE): fetch pages [startpage, startpage+npages) -- lets corpus growth pull rows
43// BEYOND an eval subset's range (disjoint corpus/eval). Same atomicity/idempotency/teeth as the base fn.
44func rf_fetch_bank_paged_from(urlbase: *u8, name: *u8, store: *TrustStore, pagesize: i64, npages: i64, startpage: i64) -> i64 {
45 let rawpath: *u8 = sys_mmap(K_MAGIC_4096); rf_bpath(rawpath, "knowledge/fetched/" as *u8, name, ".raw" as *u8)
46 let partpath: *u8 = sys_mmap(K_MAGIC_4096); rf_bpath(partpath, "knowledge/fetched/" as *u8, name, ".raw.part" as *u8)
47 if rf_have(rawpath) == 1 { rf_puts(" PAGED "); rf_puts(name); rf_puts(" [banked-skip]\n" as *u8); return 1 }
48 let fd: i64 = sys_openat_wr(partpath, 0x1a4)
49 if fd < 0 { rf_puts(" paged: cannot open .raw.part for write\n" as *u8); return 0 }
50 let cap: i64 = K_MAGIC_8388608 // 8MB per page (a 100-row page is 0.1-1MB; 500-row ~3MB)
51 let out: *u8 = sys_mmap(cap)
52 let st: *i64 = sys_mmap(16) as *i64
53 let urlbuf: *u8 = sys_mmap(K_MAGIC_4096)
54 var page: i64 = 0
55 var okpages: i64 = 0
56 var totbytes: i64 = 0
57 var failed: i64 = 0
58 while page < npages {
59 if failed == 0 {
60 var o: i64 = 0
61 o = rfx_cat(urlbuf, o, urlbase)
62 o = rfx_cat(urlbuf, o, "&offset=" as *u8)
63 o = rfx_uitoa(urlbuf, o, (startpage+page)*pagesize)
64 o = rfx_cat(urlbuf, o, "&length=" as *u8)
65 o = rfx_uitoa(urlbuf, o, pagesize)
66 urlbuf[o] = (0 as u8)
67 var n: i64 = 0
68 var att: i64 = 0
69 while att < 3 {
70 n = nx_https_fetch_follow(urlbuf, store, out, cap, 6, st)
71 if n > 0 { if st[0] == 200 { att = 3 } else { att = att + 1; rf_sleep_ms(400) } } else { att = att + 1; rf_sleep_ms(400) }
72 }
73 var banked: i64 = 0
74 if n > 0 { if n < cap { if st[0] == 200 { if rf_is_gzip(out, n) == 0 {
75 var w: i64 = 0
76 var werr: i64 = 0
77 while w < n { let ww: i64 = sys_write(fd, (out as i64 + w) as *u8, n - w); if ww <= 0 { werr = 1; w = n } else { w = w + ww } }
78 if werr == 0 { totbytes = totbytes + n; okpages = okpages + 1; banked = 1 }
79 } } } }
80 rf_puts(" page "); rf_putn(page); rf_puts(" status="); rf_putn(st[0]); rf_puts(" bytes="); rf_putn(n)
81 if banked == 1 { rf_puts(" [banked]\n" as *u8) } else { rf_puts(" [FAILED]\n" as *u8); failed = 1 }
82 rf_sleep_ms(250) // polite inter-page delay
83 }
84 page = page + 1
85 }
86 sys_close(fd)
87 rf_puts(" PAGED "); rf_puts(name); rf_puts(": "); rf_putn(okpages); rf_puts("/"); rf_putn(npages)
88 rf_puts(" pages, "); rf_putn(totbytes); rf_puts(" bytes (~"); rf_putn(okpages*pagesize); rf_puts(" rows)\n" as *u8)
89 if okpages == npages {
90 let rr: i64 = sys_renameat(partpath, rawpath)
91 if rr == 0 { if rf_have(rawpath) == 1 { rf_puts(" [promoted .part -> .raw]\n" as *u8); return 1 } }
92 rf_puts(" [RENAME FAILED -- left as .part]\n" as *u8)
93 return 0
94 }
95 rf_puts(" [INCOMPLETE -- left as .part, NOT promoted]\n" as *u8)
96 return 0
97}