nx_robust_dl.nx source

1// nx_robust_dl.nx -- the ROBUST segment-download core (the "keep downloading, no broken fragments" engine), 2// grounded in the download-SOTA corpus (yt-dlp fragment-retries, aria2 max-tries/continue, HTTP Range resume, 3// TS sync-byte integrity, exponential backoff). Format-agnostic (HLS .ts, DASH .m4s, direct) via a fetch 4// CALLBACK, so nx_hls_get / a DASH orchestrator / the API daemon all wrap it, and the gate injects failures. 5// 6// rdl_verify -- reject a truncated/corrupt fragment BEFORE it's stitched (the anti-broken-fragment gate) 7// rdl_fetch_retry -- fetch a fragment with up to max_tries attempts + integrity-verify each + backoff 8// rdl_state_* -- a persisted done-bitmap so an interrupted download RESUMES (skips done, never holes) 9import "nx_syscalls.nx" 10 11const RDL_GENERIC: i64 = 0 12const RDL_TS: i64 = 1 // MPEG-TS: 188-byte packets, each starting with the 0x47 sync byte 13 14// ---- integrity: is this fragment WHOLE + well-formed (not truncated/garbage)? ---- 15func rdl_verify_ts(bytes: *u8, len: i64) -> i64 { 16 if len <= 0 { return 0 } 17 if (len % 188) != 0 { return 0 } // a whole TS fragment is a multiple of 188 bytes 18 if (bytes[0] & 0xff) != 0x47 { return 0 } // first packet's sync byte 19 var i: i64 = 0 20 var checked: i64 = 0 21 while checked < 16 { // spot-check the first 16 packet boundaries 22 if i >= len { checked = 16 } 23 else { if (bytes[i] & 0xff) != 0x47 { return 0 } i = i + 188; checked = checked + 1 } 24 } 25 return 1 26} 27func rdl_verify(bytes: *u8, len: i64, kind: i64) -> i64 { 28 if len <= 0 { return 0 } 29 if kind == RDL_TS { return rdl_verify_ts(bytes, len) } 30 return 1 // generic (mp4/m4s/webm): non-empty is the minimum bar 31} 32 33// ---- retry: fetch one fragment reliably ---- 34// fetch_fn(url, urllen, attempt, out, outcap, rctx) -> bytes fetched (<=0 = network failure). Called with the 35// attempt index so a live fetcher can add a Range header / vary. Between attempts we back off (skipped when 36// backoff_ms<=0, e.g. in the gate). Returns the verified byte count, or -1 after exhausting max_tries -- and a 37// -1 means the caller MUST NOT stitch this fragment (that is exactly how a broken fragment gets avoided). 38func rdl_fetch_retry(url: *u8, urllen: i64, out: *u8, outcap: i64, kind: i64, 39 fetch_fn: func(*u8, i64, i64, *u8, i64, i64) -> i64, rctx: i64, 40 max_tries: i64, backoff_ms: i64) -> i64 { 41 var attempt: i64 = 0 42 while attempt < max_tries { 43 let n: i64 = fetch_fn(url, urllen, attempt, out, outcap, rctx) 44 if n > 0 { if rdl_verify(out, n, kind) == 1 { return n } } // got bytes AND they're whole -> done 45 // network failure OR integrity failure -> exponential backoff, then retry 46 if backoff_ms > 0 { var d: i64 = backoff_ms; var s: i64 = 0; while s < attempt { d = d * 2; s = s + 1 } sys_sleep_ms(d) } 47 attempt = attempt + 1 48 } 49 return 0 - 1 50} 51 52// ---- resume: a done-bitmap persisted so an interrupted run picks up where it stopped ---- 53// state is a byte-per-segment array (state[idx]=1 => that segment is already downloaded+verified). On resume 54// the driver skips done segments (no re-download) and only the missing ones are fetched (no holes). 55func rdl_state_mark(state: *u8, idx: i64) -> i64 { state[idx] = 1 as u8; return 0 } 56func rdl_state_done(state: *u8, idx: i64) -> i64 { if (state[idx] & 0xff) == 1 { return 1 } return 0 } 57func rdl_state_count(state: *u8, n: i64) -> i64 { var c: i64 = 0; var i: i64 = 0; while i < n { if (state[i] & 0xff) == 1 { c = c + 1 } i = i + 1 } return c } 58// persist the done-bitmap to `path` (n bytes). Atomic-ish: caller can write to .tmp then rename in prod. 59func rdl_state_save(path: *u8, state: *u8, n: i64) -> i64 { 60 let fd: i64 = sys_openat_wr(path, 0x1a4) 61 if fd < 0 { return 0 - 1 } 62 let w: i64 = sys_write(fd, state, n) 63 sys_close(fd) 64 if w != n { return 0 - 1 } 65 return 0 66} 67// load the done-bitmap from `path` into state (n bytes); returns bytes read, or 0 if absent (fresh download). 68func rdl_state_load(path: *u8, state: *u8, n: i64) -> i64 { 69 let fd: i64 = sys_openat_rd(path) 70 if fd < 0 { var i: i64 = 0; while i < n { state[i] = 0 as u8; i = i + 1 } return 0 } 71 let rd: i64 = sys_read(fd, state, n) 72 sys_close(fd) 73 return rd 74}