nx_fetch_deadline_candidate_t314.nx source
↩ module page · 195 lines · 8458 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}
147
148// Caller-owned nonblocking fd. One absolute same-host monotonic deadline for all I/O.
149// Does not silently convert an expired budget into the legacy default.
150func fb_wait_event_until(fd:i64,events:i64,deadline_ms:i64)->i64 {
151 var pf:i64=0
152 while 1==1 {
153 let now:i64=sys_now_ms()
154 if now>=deadline_ms { return FB_TIMEOUT }
155 let left:i64=deadline_ms-now
156 fb_pfd_set((&pf) as *u8,fd,events)
157 let rc:i64=sys_poll((&pf) as *u8,1,left)
158 if rc==0 { return FB_TIMEOUT }
159 if rc>0 { return FB_OK }
160 if rc!=(0-4) { return FB_READ_ERR }
161 }
162 return FB_READ_ERR
163}
164func fb_send_until(fd:i64,src:*u8,n:i64,deadline_ms:i64,written:*i64)->i64 {
165 written[0]=0
166 if n<0 { return FB_READ_ERR }
167 while written[0]<n {
168 let ready:i64=fb_wait_event_until(fd,4,deadline_ms)
169 if ready!=FB_OK { return ready }
170 let rc:i64=sys_sendto(fd,src+written[0],n-written[0],16384,0 as *u8,0)
171 if rc>0 { written[0]=written[0]+rc } else {
172 if rc!=(0-4) { if rc!=(0-11) { return FB_READ_ERR } }
173 }
174 }
175 return FB_OK
176}
177func fb_receive_until(fd:i64,out:*u8,cap:i64,deadline_ms:i64,counted:*i64)->i64 {
178 counted[0]=0
179 if cap<0 { return FB_READ_ERR }
180 var extra:i64=0
181 while 1==1 {
182 let ready:i64=fb_wait_event_until(fd,1,deadline_ms)
183 if ready!=FB_OK { return ready }
184 var dst:*u8=(&extra) as *u8
185 var room:i64=1
186 if counted[0]<cap { dst=out+counted[0];room=cap-counted[0] }
187 let rc:i64=sys_read(fd,dst,room)
188 if rc==0 { return FB_OK }
189 if rc>0 {
190 if counted[0]==cap { return FB_OVERFLOW }
191 counted[0]=counted[0]+rc
192 } else { if rc!=(0-4) { if rc!=(0-11) { return FB_READ_ERR } } }
193 }
194 return FB_READ_ERR
195}