code wiki / (root) / nx_fetch_bounded.nx

nx_fetch_bounded.nx source

↩ module page · 146 lines · 6663 B

1// nx_fetch_bounded.nx -- bounded + timed streaming fetch drain (RES-R7). 2// 3// ROOT CAUSE this organ fixes: 4// The existing drains silently lose data and can hang forever. 5// - nx_browse_text.nx:55 br_drain : fills to `cap`, returns `off`, 6// gives the caller NO way to tell "exactly cap" from "truncated at 7// cap" -- silent truncation. No read timeout -> a stalled peer 8// blocks sys_read forever. 9// - nx_fetch_unit.nx:172 : `if blen > cap { blen = cap }` 10// -- a SILENT clamp: oversize bodies are quietly chopped, length 11// lost. No per-fetch timeout. 12// 13// THE FIX (mirrors the h2 large-page streaming-consume contract, 14// h2_stream_consume): count the wire length UNBOUNDED, copy only up to 15// `cap`, and report BOTH an explicit status code AND the full counted 16// length. Never silently clamp. Enforce a wall-clock deadline via 17// sys_poll so a stalled read returns within the budget with a timeout 18// code instead of hanging the conductor. 19// 20// SINGLE RESPONSIBILITY: bounded+timed streaming drain over ANY readable 21// fd. Composes only nx_syscalls.nx primitives (sys_poll, sys_read, 22// sys_now_us, sys_mmap) -- fully sovereign, no network/TLS/gcc. The 23// existing br_drain / nx_fetch_staged are NOT modified (additive-only); 24// callers migrate to fb_drain_bounded in a later rung. 25// 26// build/placement: harness resolves `nx_fetch_bounded` -> 27// runtime/nx_fetch_bounded.nx (2nd fallback). 28// expect_exit: 0 29// license_tier: ORIGINAL 30 31import "nx_syscalls.nx" 32 33// ===== Result codes ================================================= 34// Returned by fb_drain_bounded. The companion `counted_out` ALWAYS 35// receives the true unbounded byte count, so the caller never has to 36// guess whether `cap` meant "exactly cap" or "truncated at cap". 37const FB_OK: i64 = 0 // fully consumed within cap and before deadline 38const FB_OVERFLOW: i64 = 0 - 70 // body exceeded cap; full length still counted (no silent trunc) 39const FB_TIMEOUT: i64 = 0 - 71 // deadline hit before EOF (slow / stalled peer) 40const FB_READ_ERR: i64 = 0 - 72 // sys_read returned < 0 (not EOF) 41 42// ===== Config-style constants (no magic numbers inline) ============= 43const FB_DEFAULT_TIMEOUT_MS: i64 = 30000 // per-fetch wall budget when caller passes <= 0 44const FB_READ_SLICE: i64 = 65536 // streaming read granularity (compact-as-you-go) 45const FB_POLLIN: i64 = 1 // POLLIN event mask (matches nx_syscalls POLLIN) 46 47// ===== pollfd helper (lifted idiom from nx_route_probe.nx:47-54) ==== 48// struct pollfd { i32 fd; i16 events; i16 revents } -- 8 bytes. 49func fb_pfd_set(pf: *u8, fd: i64, events: i64) -> i64 { 50 pf[0] = (fd & 0xff) as u8 51 pf[1] = ((fd >> 8) & 0xff) as u8 52 pf[2] = ((fd >> 16) & 0xff) as u8 53 pf[3] = ((fd >> 24) & 0xff) as u8 54 pf[4] = (events & 0xff) as u8 55 pf[5] = ((events >> 8) & 0xff) as u8 56 pf[6] = 0 as u8 57 pf[7] = 0 as u8 58 return 0 59} 60 61// Wait until `fd` is readable or `deadline_us` passes. Returns: 62// 1 -> fd readable (sys_poll reported a ready fd) 63// 0 -> timeout (deadline reached with no data) 64// <0 -> poll error (-errno) 65// Computes the REMAINING budget each call so repeated waits across a 66// streaming loop still honour ONE overall deadline (not per-iteration). 67func fb_wait_readable(fd: i64, deadline_us: i64) -> i64 { 68 let now: i64 = sys_now_us() 69 if now >= deadline_us { return 0 } // already past the deadline 70 let remain_us: i64 = deadline_us - now 71 var remain_ms: i64 = remain_us / 1000 72 if remain_ms <= 0 { remain_ms = 1 } // poll(0) returns immediately; give >=1ms 73 let pf: *u8 = sys_mmap(8) 74 fb_pfd_set(pf, fd, FB_POLLIN) 75 let r: i64 = sys_poll(pf, 1, remain_ms) 76 if r < 0 { return r } 77 if r == 0 { return 0 } 78 return 1 79} 80 81// THE capability: stream-consume `fd` into out[0..cap), enforcing a 82// single wall-clock deadline. 83// out : caller buffer; bytes BEYOND cap are COUNTED, not stored 84// cap : capacity of `out` 85// timeout_ms : per-fetch budget; <= 0 means FB_DEFAULT_TIMEOUT_MS 86// counted_out : *i64 <- total bytes seen on the wire (UNBOUNDED) 87// returns : FB_OK | FB_OVERFLOW | FB_TIMEOUT | FB_READ_ERR 88// 89// Loop invariant: `written` <= cap at all times; `*counted_out` grows 90// by every byte read. On EOF, overflow == (counted > cap). A stalled 91// peer trips fb_wait_readable -> FB_TIMEOUT. No silent clamp anywhere. 92func fb_drain_bounded(fd: i64, out: *u8, cap: i64, timeout_ms: i64, counted_out: *i64) -> i64 { 93 var budget_ms: i64 = timeout_ms 94 if budget_ms <= 0 { budget_ms = FB_DEFAULT_TIMEOUT_MS } 95 let deadline_us: i64 = sys_now_us() + budget_ms * 1000 96 97 let slice: *u8 = sys_mmap(FB_READ_SLICE) 98 var counted: i64 = 0 // unbounded wire length 99 var written: i64 = 0 // bytes actually stored in out (<= cap) 100 var status: i64 = FB_OK 101 var running: i64 = 1 102 103 while running == 1 { 104 let ready: i64 = fb_wait_readable(fd, deadline_us) 105 if ready == 0 { status = FB_TIMEOUT; running = 0 } 106 if running == 1 { 107 if ready < 0 { status = FB_READ_ERR; running = 0 } 108 } 109 if running == 1 { 110 let r: i64 = sys_read(fd, slice, FB_READ_SLICE) 111 if r < 0 { status = FB_READ_ERR; running = 0 } 112 if running == 1 { 113 if r == 0 { 114 running = 0 // EOF -- clean end of stream 115 } else { 116 counted = counted + r // ALWAYS count full wire length 117 // compact-as-you-go: copy only what still fits in cap 118 var i: i64 = 0 119 while i < r { 120 if written < cap { 121 out[written] = slice[i] 122 written = written + 1 123 } 124 i = i + 1 125 } 126 } 127 } 128 } 129 } 130 131 counted_out[0] = counted 132 if status == FB_TIMEOUT { return FB_TIMEOUT } 133 if status == FB_READ_ERR { return FB_READ_ERR } 134 if counted > cap { return FB_OVERFLOW } // honest reject, full length reported 135 return FB_OK 136} 137 138// ===== scaffold main ================================================ 139// Library organ: no socket work in main (single responsibility). A 140// trivial main lets it build standalone via the harness (mirrors 141// nx_h2_serve.nx:241). The REAL proof lives in the gate organ. 142func main() -> i64 { 143 sys_write(1, "nx_fetch_bounded: library organ (fb_drain_bounded) -- see _fetch_bounded_gate\n" as *u8, 77) 144 sys_exit(0) 145 return 0 146}