nx_apertus_shards_pull.nx source

1// nx_apertus_shards_pull.nx -- S-CLASS resumable, self-healing shard pull (operator: max speed / resume / error 2// handling). Streams each shard to the NAS via nx_https_get_stream (one TLS conn, no buffering), and: 3// RESUME -- before each attempt, lseek the .part to its current size and request "Range: bytes=<size>-" so a 4// broken/killed download CONTINUES from where it stopped (never re-fetches the bytes already on disk). 5// RETRY -- wraps each shard in an attempt loop: on any error (NAS write fail, read timeout, connection drop) it 6// reconnects and resumes; progress resets the try counter (self-healing), only true stalls exhaust it. 7// SAFE -- if a server ignores Range and returns a full 200, get_stream returns -9 and we truncate + restart 8// (never append a full body onto a partial). Final integrity is still checked by nx_apertus_shard_verify. 9// Reuses nx_https_fetch_follow's plumbing (ff_path/ff_resolve_location/url_connect/session_run). license_tier: ORIGINAL 10import "nx_syscalls.nx" 11import "nx_csprng.nx" 12import "nx_trust_store_load_from_certdata.nx" 13import "nx_https_fetch_follow.nx" 14import "nx_https_get_stream.nx" 15const K_MAGIC_4096: i64 = 4096 16const K_MAGIC_2048: i64 = 2048 17const K_MAGIC_4194304: i64 = 4194304 18const K_MAGIC_4999776656: i64 = 4999776656 19const K_MAGIC_4882374192: i64 = 4882374192 20const K_MAGIC_4974647808: i64 = 4974647808 21const K_MAGIC_1249928160: i64 = 1249928160 22const K_MAGIC_1024: i64 = 1024 23 24func w(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} sys_write(1,s,n); return 0 } 25func wn(v: i64) -> i64 { var m: i64=v; if m<0{w("-" as *u8);m=0-m} let t:*u8=sys_mmap(24); var k:i64=0; if m==0{t[0]=48 as u8;k=1} while m>0{t[k]=(48+(m%10)) as u8;m=m/10;k=k+1} var i:i64=0; let o:*u8=sys_mmap(24); while i<k{o[i]=t[k-1-i];i=i+1} sys_write(1,o,k); return 0 } 26func sp_cat(dst: *u8, off: i64, s: *u8) -> i64 { var o: i64=off; var i: i64=0; while s[i]!=(0 as u8){dst[o]=s[i];o=o+1;i=i+1} return o } 27 28// current size of a file (0 if absent) via lseek SEEK_END. 29func file_size(path: *u8) -> i64 { 30 let fd: i64 = sys_openat_rd(path) 31 if fd < 0 { return 0 } 32 let sz: i64 = sys_lseek(fd, 0, 2) 33 sys_close(fd) 34 if sz < 0 { return 0 } 35 return sz 36} 37 38// ONE attempt: open dest (append if resuming, truncate if fresh), hop-loop connect+handshake, stream (Range=have). 39// Returns get_stream's final rc: >0 body bytes this attempt, -9 range-ignored (caller truncates+restarts), <0 error. 40func pull_attempt(url0: *u8, store: *TrustStore, dest_path: *u8, have: i64) -> i64 { 41 let urlbuf: *u8 = sys_mmap(K_MAGIC_4096) 42 var ui: i64 = 0; while url0[ui] != 0 as u8 { urlbuf[ui] = url0[ui]; ui = ui + 1 } urlbuf[ui] = 0 as u8 43 44 var dest_fd: i64 = 0 - 1 45 if have > 0 { dest_fd = sys_openat_append(dest_path, 0x1a4) } else { dest_fd = sys_openat_wr(dest_path, 0x1a4) } 46 if dest_fd < 0 { return 0 - 100 } 47 48 let cr: *u8 = sys_mmap(32) 49 let priv: *u8 = sys_mmap(32) 50 let statusp: *i64 = sys_mmap(16) as *i64 51 let loc: *u8 = sys_mmap(K_MAGIC_4096) 52 53 var hop: i64 = 0 54 while hop <= 8 { 55 let target_raw: *u8 = sys_mmap(64) 56 let target: *NxHttpsTarget = target_raw as *NxHttpsTarget 57 target.url = nx_url_new() 58 target.port = 0 59 if nx_https_url_for_fetch(urlbuf, target) != NX_HTTPS_URL_OK { sys_close(dest_fd); return 0 - 101 } 60 let fd_p: *i64 = (sys_mmap(16)) as *i64 61 if nx_https_url_connect(target, urlbuf, sys_now_realtime_sec(), fd_p) != NX_HTTPS_CONNECT_OK { sys_close(dest_fd); return 0 - 102 } 62 let fd: i64 = fd_p[0] 63 var i: i64 = 0 64 nx_csprng_fill(cr, 32); nx_csprng_fill(priv, 32) // CWE-330 (debt 1785970852): were the constants 0xC0../0xA0.. on EVERY session 65 let vc_raw: *u8 = sys_mmap(64) 66 let vc: *TlsValidationContext = vc_raw as *TlsValidationContext 67 vc.store = store 68 vc.sni_host = urlbuf + target.url.host_off 69 vc.sni_host_len = target.url.host_len 70 vc.now_epoch = sys_now_realtime_sec() 71 let sr: i64 = nx_tls13_client_session_run(fd, urlbuf + target.url.host_off, target.url.host_len, cr, priv, vc) 72 if sr <= 0 { sys_close(fd); sys_close(dest_fd); return 0 - 103 } 73 let session: *Tls13ClientSession = sr as *Tls13ClientSession 74 let path: *u8 = sys_mmap(K_MAGIC_2048) 75 let plen: i64 = ff_path(urlbuf, target, path) 76 let rc: i64 = nx_https_get_stream(session, fd, path, plen, urlbuf + target.url.host_off, target.url.host_len, have, dest_fd, statusp, loc, K_MAGIC_4096) 77 sys_close(fd) 78 if rc == 0 { 79 // redirect: resolve Location -> urlbuf, next hop (dest untouched on 3xx) 80 let resolved: *u8 = sys_mmap(K_MAGIC_4096) 81 ff_resolve_location(loc, urlbuf, target, resolved) 82 var k: i64 = 0; while resolved[k] != 0 as u8 { urlbuf[k] = resolved[k]; k = k + 1 } urlbuf[k] = 0 as u8 83 hop = hop + 1 84 } else { sys_close(dest_fd); return rc } // rc>0 body, -9 range-ignored, or <0 error 85 } 86 sys_close(dest_fd); return 0 - 200 87} 88 89// RESUMABLE + self-healing: retry until the .part reaches `target` (or the no-progress budget is exhausted). 90func pull_resumable(url0: *u8, store: *TrustStore, dest_path: *u8, target: i64) -> i64 { 91 var tries: i64 = 0 92 while tries < 20 { 93 let have: i64 = file_size(dest_path) 94 if have >= target { return have } 95 w(" attempt from byte " as *u8); wn(have); w(" / " as *u8); wn(target); w(" ...\n" as *u8) 96 let rc: i64 = pull_attempt(url0, store, dest_path, have) 97 if rc == (0 - 9) { 98 let tf: i64 = sys_openat_wr(dest_path, 0x1a4); if tf >= 0 { sys_close(tf) } // Range ignored -> truncate, restart 99 tries = tries + 1 100 w(" (server ignored Range -> truncated, restarting)\n" as *u8) 101 } else { 102 let have2: i64 = file_size(dest_path) 103 w(" attempt rc=" as *u8); wn(rc); w(" now at " as *u8); wn(have2); w("\n" as *u8) 104 if have2 >= target { return have2 } 105 if have2 > have { tries = 0 } else { tries = tries + 1 } // progress = self-heal; stall = count down 106 } 107 } 108 return 0 - 1 109} 110 111func main() -> i64 { 112 w("=== nx_apertus_shards_pull: RESUMABLE self-healing stream -> NAS ===\n" as *u8) 113 let r: i64 = nx_trust_store_load_from_certdata("data/mozilla_certdata.txt" as *u8, 512, K_MAGIC_4194304) 114 if r <= 0 { w("certdata load FAILED\n" as *u8); sys_exit(1); return 1 } 115 let store: *TrustStore = r as *TrustStore 116 let base: *u8 = "https://huggingface.co/adamo1139/Apertus-8B-Instruct-2509-ungated/resolve/main/" as *u8 117 let destbase: *u8 = "/mnt/nas_ai/apertus/" as *u8 118 119 let names: *i64 = sys_mmap(8 * 8) as *i64 120 let sizes: *i64 = sys_mmap(8 * 8) as *i64 121 names[0] = "model-00001-of-00004.safetensors" as *u8 as i64; sizes[0] = K_MAGIC_4999776656 122 names[1] = "model-00002-of-00004.safetensors" as *u8 as i64; sizes[1] = K_MAGIC_4882374192 123 names[2] = "model-00003-of-00004.safetensors" as *u8 as i64; sizes[2] = K_MAGIC_4974647808 124 names[3] = "model-00004-of-00004.safetensors" as *u8 as i64; sizes[3] = K_MAGIC_1249928160 125 let nn: i64 = 4 126 127 var ok: i64 = 0 128 var i: i64 = 0 129 while i < nn { 130 let name: *u8 = names[i] as *u8 131 let target: i64 = sizes[i] 132 let url: *u8 = sys_mmap(K_MAGIC_1024); var uo: i64 = sp_cat(url, 0, base); uo = sp_cat(url, uo, name); url[uo] = 0 as u8 133 let dest: *u8 = sys_mmap(K_MAGIC_1024); var doff: i64 = sp_cat(dest, 0, destbase); doff = sp_cat(dest, doff, name); doff = sp_cat(dest, doff, ".part" as *u8); dest[doff] = 0 as u8 134 w(" " as *u8); w(name); w(" (target=" as *u8); wn(target); w(")\n" as *u8) 135 let fin: i64 = pull_resumable(url, store, dest, target) 136 if fin >= target { w(" COMPLETE: " as *u8); wn(fin); w(" bytes on NAS\n" as *u8); ok = ok + 1 } else { w(" INCOMPLETE: reached " as *u8); wn(fin); w(" / " as *u8); wn(target); w("\n" as *u8) } 137 i = i + 1 138 } 139 w("\n SHARDS COMPLETE: " as *u8); wn(ok); w("/" as *u8); wn(nn); w("\n" as *u8) 140 if ok == nn { w("=== GREEN (all 4 shards fully on the NAS; run nx_apertus_shard_verify to promote) ===\n" as *u8); sys_exit(0); return 0 } 141 w("=== PARTIAL (re-run to resume the rest) ===\n" as *u8); sys_exit(1); return 1 142}