tracker

57.9 KB

1. // hl:fetch — the HTTP(S) CLIENT, as a server-realm plugin (mission 135).
2. //
3. // Shaped on the web `fetch` the creator already knows: a method, a url, headers and
4. // a body go out; a status, headers and a body come back. What is NOT web-shaped is
5. // where the wait happens, and that is the one design decision in this file:
6. //
7. // `fetch.start` NEVER WAITS. It hands the request to a thread of its own and
8. // returns a TICKET (an HlHandle) immediately. `ticket.result()` is the wait.
9. //
10. // That split is what makes `concurrent [ fetchStart(a), fetchStart(b), … ]` put N
11. // requests on the wire AT ONCE: a `concurrent` arm runs to completion before its
12. // sibling starts (nothing in the interpreter yields today), so an arm that blocked
13. // on the socket would serialize the whole group. An arm that only starts a thread
14. // does not. `fetch()` in server.hl is `fetchStart(…).response()` — start plus wait,
15. // the familiar one-liner, for code that has nothing else to do.
16. //
17. // THE EVENT LOOP is never blocked by a fetch in flight: no loop thread is inside a
18. // socket call, and a completed request rings the mission-125 BELL — `fetch.completions`
19. // is an ordinary loop source with a `wake_fd`, so an `on fetched(ev)` handler is woken
20. // by the same `epoll_wait` that wakes an http1 request. (The wait in `result()` is a
21. // wait like any other plugin call's: it blocks the FIBER that asked. Starting the
22. // fetches early and reading them later is what keeps a server serving, and the
23. // completions source is how a server never has to wait at all.)
24. //
25. // TLS is `plugins/http/tls_common.zig` — the same module hl:http1 and hl:http2 use,
26. // extended with a client half in this mission. Verification is ON: system trust
27. // store plus hostname checking, with `caFile` naming an EXTRA anchor for a test
28. // fixture's self-signed certificate. There is no verify-off switch at all.
29. //
30. // HTTP/1.1 only. An h2 client is a later slice: it needs ALPN on the client side and
31. // a multiplexed connection pool, neither of which this file has. `connection: close`
32. // on every request, so there is no pool to keep coherent either.
34. const std = @import("std");
35. const api = @import("plugin_api");
36. const http = @import("http_common");
37. const tls = @import("tls_common");
39. const HlValue = api.HlValue;
40. const HlField = api.HlField;
41. const HlObject = api.HlObject;
42. const HlIterator = api.HlIterator;
43. const HlHandle = api.HlHandle;
45. const c = @cImport({
46. @cInclude("netdb.h");
47. @cInclude("sys/socket.h");
48. @cInclude("sys/time.h");
49. @cInclude("unistd.h");
50. @cInclude("netinet/in.h");
51. @cInclude("netinet/tcp.h");
52. });
54. // requests run on their own threads: the plugins' allocator (plugin_api.zig)
55. const allocator = api.allocator;
57. const linux = std.os.linux;
58. const libc = std.c;
60. // pthread rather than std.Thread.Mutex, exactly as hl:http1 does: this is a `.so`
61. // loaded into a binary built by a possibly different Zig, and the C primitives are
62. // the ones whose ABI is fixed by the platform rather than by the standard library.
63. const PthreadMutex = libc.pthread_mutex_t;
64. const PthreadCond = libc.pthread_cond_t;
66. fn mutexLock(m: *PthreadMutex) void {
67. _ = libc.pthread_mutex_lock(m);
68. }
69. fn mutexUnlock(m: *PthreadMutex) void {
70. _ = libc.pthread_mutex_unlock(m);
71. }
72. fn condBroadcast(cnd: *PthreadCond) void {
73. _ = libc.pthread_cond_broadcast(cnd);
74. }
75. fn condWait(cnd: *PthreadCond, m: *PthreadMutex) void {
76. _ = libc.pthread_cond_wait(cnd, m);
77. }
79. fn logMsg(msg: []const u8) void {
80. _ = linux.write(2, msg.ptr, msg.len);
81. }
83. fn logFmt(comptime fmt: []const u8, args: anytype) void {
84. var buf: [512]u8 = undefined;
85. const s = std.fmt.bufPrint(&buf, fmt, args) catch return;
86. logMsg(s);
87. }
89. // ── Limits, all of them documented defaults rather than hard walls ───────────
90. /// No answer at all within this many milliseconds is a failed fetch. Overridable
91. /// per call with `timeoutMs`; it covers the WHOLE exchange (DNS, connect,
92. /// handshake, request, response, and every redirect hop), not one syscall.
93. const DEFAULT_TIMEOUT_MS: i64 = 30_000;
94. /// How many 3xx hops are followed before the fetch fails with "too many redirects".
95. /// Overridable with `maxRedirects`; 0 means the 3xx is RETURNED as the response.
96. const DEFAULT_MAX_REDIRECTS: u32 = 5;
97. /// A response body larger than this fails rather than growing the process without
98. /// bound — a client that pulls other people's APIs must not be a memory bomb.
99. const MAX_BODY: usize = 32 * 1024 * 1024;
101. /// O_NONBLOCK on x86-64 Linux. Spelled here because <fcntl.h> cannot be imported
102. /// (see tcpConnect); it is an ABI constant, not a libc detail.
103. const O_NONBLOCK: usize = 0o4000;
105. const Header = struct {
106. name: []u8,
107. value: []u8,
108. };
110. fn freeHeaders(list: []Header) void {
111. for (list) |h| {
112. allocator.free(h.name);
113. allocator.free(h.value);
114. }
115. allocator.free(list);
116. }
118. // ── The request, as the ticket carries it ────────────────────────────────────
120. const Request = struct {
121. method: []u8,
122. url: []u8,
123. headers: []Header,
124. body: []u8,
125. has_body: bool,
126. json_body: bool,
127. timeout_ms: i64,
128. max_redirects: u32,
129. ca_file: ?[]u8,
130. tag: []u8,
131. /// STREAMING (the LLM-token mode): deliver the body INCREMENTALLY as chunk
132. /// events on the completions source instead of accumulating it. The final
133. /// completion then carries status/headers and an EMPTY body — the chunks
134. /// were the body. Requires an armed completions source; `result()` still
135. /// works and yields the empty-body summary when the stream ends.
136. stream: bool = false,
138. fn deinit(self: *Request) void {
139. allocator.free(self.method);
140. allocator.free(self.url);
141. freeHeaders(self.headers);
142. allocator.free(self.body);
143. if (self.ca_file) |ca| allocator.free(ca);
144. allocator.free(self.tag);
145. }
146. };
148. // ── The ticket: one in-flight (or finished) request ──────────────────────────
150. const Ticket = struct {
151. mutex: PthreadMutex = libc.PTHREAD_MUTEX_INITIALIZER,
152. cond: PthreadCond = libc.PTHREAD_COND_INITIALIZER,
153. done: bool = false,
154. thread: ?std.Thread = null,
156. req: Request,
158. status: u16 = 0,
159. final_url: []u8 = &.{},
160. headers: []Header = &.{},
161. body: []u8 = &.{},
162. /// Non-null = the fetch failed and this is the located message the handle
163. /// hands back as an `hl_error`.
164. err: ?[]u8 = null,
165. /// ABORT (the LLM-interrupt primitive): the socket currently carrying this
166. /// request, visible under the ticket mutex so `fetchAbort(tag)` can
167. /// shutdown() it from the interpreter thread — the read unblocks, the peer
168. /// sees the disconnect and stops generating. -1 = none in flight.
169. live_fd: i32 = -1,
170. aborted: bool = false,
171. /// INSTANCE EVENTS (stream mode): a streaming ticket carries its OWN frame
172. /// queue and bell, created at start — before the worker exists — so no chunk
173. /// can outrun the arming. `handle.events()` wraps them into a loop source;
174. /// pending_fetch.hl registers it and re-emits `chunk`/`done` at itself.
175. /// -1 = not a streaming ticket (completions go the global route, if armed).
176. ev_queue: std.ArrayListUnmanaged(Completion) = .empty,
177. ev_wake_fd: i32 = -1,
179. fn wait(self: *Ticket) void {
180. mutexLock(&self.mutex);
181. while (!self.done) condWait(&self.cond, &self.mutex);
182. mutexUnlock(&self.mutex);
183. }
185. fn isDone(self: *Ticket) bool {
186. mutexLock(&self.mutex);
187. defer mutexUnlock(&self.mutex);
188. return self.done;
189. }
191. fn setLiveFd(self: *Ticket, fd: i32) bool {
192. mutexLock(&self.mutex);
193. defer mutexUnlock(&self.mutex);
194. if (self.aborted) return false; // aborted before the connect landed
195. self.live_fd = fd;
196. return true;
197. }
199. fn clearLiveFd(self: *Ticket) void {
200. mutexLock(&self.mutex);
201. self.live_fd = -1;
202. mutexUnlock(&self.mutex);
203. }
205. /// The abort itself: mark, and shutdown() any in-flight socket so the
206. /// worker's blocking read returns NOW. Closing is still the worker's job
207. /// (its defer owns the fd); shutdown only cuts the conversation.
208. fn abort(self: *Ticket) void {
209. mutexLock(&self.mutex);
210. self.aborted = true;
211. if (self.live_fd >= 0) _ = c.shutdown(self.live_fd, 2); // SHUT_RDWR
212. mutexUnlock(&self.mutex);
213. }
215. fn finish(self: *Ticket) void {
216. mutexLock(&self.mutex);
217. self.done = true;
218. condBroadcast(&self.cond);
219. mutexUnlock(&self.mutex);
220. }
222. fn fail(self: *Ticket, comptime fmt: []const u8, args: anytype) void {
223. self.err = std.fmt.allocPrint(allocator, fmt, args) catch null;
224. }
225. };
227. // ── URL ──────────────────────────────────────────────────────────────────────
229. const Url = struct {
230. https: bool,
231. host: []const u8,
232. port: u16,
233. /// path + query, always starting with '/'
234. target: []const u8,
235. };
237. /// `scheme://host[:port][/path][?query]`. Only http and https; a URL with
238. /// credentials or an IPv6 literal is refused rather than half-understood.
239. fn parseUrl(url: []const u8) ?Url {
240. var rest: []const u8 = undefined;
241. var https = false;
242. if (std.mem.startsWith(u8, url, "http://")) {
243. rest = url["http://".len..];
244. } else if (std.mem.startsWith(u8, url, "https://")) {
245. rest = url["https://".len..];
246. https = true;
247. } else return null;
249. const slash = std.mem.indexOfScalar(u8, rest, '/');
250. const authority = if (slash) |s| rest[0..s] else rest;
251. const target = if (slash) |s| rest[s..] else "/";
252. if (authority.len == 0) return null;
253. if (std.mem.indexOfScalar(u8, authority, '@') != null) return null;
254. if (std.mem.indexOfScalar(u8, authority, '[') != null) return null;
256. var host = authority;
257. var port: u16 = if (https) 443 else 80;
258. if (std.mem.lastIndexOfScalar(u8, authority, ':')) |colon| {
259. host = authority[0..colon];
260. port = std.fmt.parseInt(u16, authority[colon + 1 ..], 10) catch return null;
261. }
262. if (host.len == 0) return null;
264. return .{ .https = https, .host = host, .port = port, .target = target };
265. }
267. /// Resolve a `Location` against the URL it came from: absolute stays, `/x` replaces
268. /// the path, anything else is relative to the current directory of the path.
269. fn resolveLocation(base: []const u8, loc: []const u8) ?[]u8 {
270. if (loc.len == 0) return null;
271. if (std.mem.startsWith(u8, loc, "http://") or std.mem.startsWith(u8, loc, "https://")) {
272. return allocator.dupe(u8, loc) catch null;
273. }
274. const b = parseUrl(base) orelse return null;
275. const scheme = if (b.https) "https" else "http";
276. const default_port: u16 = if (b.https) 443 else 80;
278. var authority_buf: [300]u8 = undefined;
279. const authority = if (b.port == default_port)
280. std.fmt.bufPrint(&authority_buf, "{s}", .{b.host}) catch return null
281. else
282. std.fmt.bufPrint(&authority_buf, "{s}:{d}", .{ b.host, b.port }) catch return null;
284. if (loc[0] == '/') {
285. return std.fmt.allocPrint(allocator, "{s}://{s}{s}", .{ scheme, authority, loc }) catch null;
286. }
287. // Relative: cut the base target back to its last '/'.
288. const path_only = if (std.mem.indexOfScalar(u8, b.target, '?')) |q| b.target[0..q] else b.target;
289. const cut = std.mem.lastIndexOfScalar(u8, path_only, '/') orelse 0;
290. return std.fmt.allocPrint(allocator, "{s}://{s}{s}/{s}", .{ scheme, authority, path_only[0..cut], loc }) catch null;
291. }
293. // ── Socket plumbing ──────────────────────────────────────────────────────────
295. fn nowMs() i64 {
296. var ts: linux.timespec = undefined;
297. _ = linux.clock_gettime(linux.CLOCK.MONOTONIC, &ts);
298. return @as(i64, ts.sec) * 1000 + @divTrunc(@as(i64, ts.nsec), 1_000_000);
299. }
301. /// Connect to host:port, giving up after `budget_ms`. The connect is non-blocking
302. /// plus a poll so the DEADLINE is honoured even when the peer never answers SYN;
303. /// the socket is put back into blocking mode with SO_RCVTIMEO/SO_SNDTIMEO after,
304. /// which is what bounds the read side.
305. fn tcpConnect(host: []const u8, port: u16, budget_ms: i64, err_out: *[]const u8) ?i32 {
306. if (budget_ms <= 0) {
307. err_out.* = "timed out";
308. return null;
309. }
311. var host_buf: [256]u8 = undefined;
312. if (host.len >= host_buf.len) {
313. err_out.* = "host name too long";
314. return null;
315. }
316. @memcpy(host_buf[0..host.len], host);
317. host_buf[host.len] = 0;
319. var port_buf: [8]u8 = undefined;
320. const port_str = std.fmt.bufPrintZ(&port_buf, "{d}", .{port}) catch {
321. err_out.* = "bad port";
322. return null;
323. };
325. var hints: c.struct_addrinfo = std.mem.zeroes(c.struct_addrinfo);
326. hints.ai_family = c.AF_UNSPEC;
327. hints.ai_socktype = c.SOCK_STREAM;
328. hints.ai_protocol = c.IPPROTO_TCP;
330. var res: ?*c.struct_addrinfo = null;
331. if (c.getaddrinfo(@ptrCast(&host_buf), port_str.ptr, &hints, &res) != 0 or res == null) {
332. err_out.* = "could not resolve host";
333. return null;
334. }
335. defer c.freeaddrinfo(res);
337. const deadline = nowMs() + budget_ms;
338. var last: []const u8 = "connection failed";
340. var it: ?*c.struct_addrinfo = res;
341. while (it) |ai| : (it = ai.ai_next) {
342. const remaining = deadline - nowMs();
343. if (remaining <= 0) {
344. err_out.* = "timed out";
345. return null;
346. }
348. const fd_c = c.socket(ai.ai_family, ai.ai_socktype, ai.ai_protocol);
349. if (fd_c < 0) {
350. last = "could not create a socket";
351. continue;
352. }
353. const fd: i32 = @intCast(fd_c);
355. // fcntl straight from the kernel: glibc's <fcntl.h> does not survive
356. // translate-c under _FORTIFY_SOURCE (its open() overloads are declared
357. // with attribute error), and the flags are ABI constants either way.
358. const flags = linux.fcntl(fd, linux.F.GETFL, 0);
359. _ = linux.fcntl(fd, linux.F.SETFL, flags | @as(usize, O_NONBLOCK));
361. var connected = false;
362. if (c.connect(fd, ai.ai_addr, ai.ai_addrlen) == 0) {
363. connected = true;
364. } else {
365. // linux.poll, not glibc's: <poll.h> is fortified too (see above).
366. var pfd = [_]linux.pollfd{.{ .fd = fd, .events = linux.POLL.OUT, .revents = 0 }};
367. const pr: isize = @bitCast(linux.poll(&pfd, 1, @intCast(@min(remaining, std.math.maxInt(i32)))));
368. if (pr > 0) {
369. var so_err: c_int = 0;
370. var len: c.socklen_t = @sizeOf(c_int);
371. _ = c.getsockopt(fd, c.SOL_SOCKET, c.SO_ERROR, &so_err, &len);
372. if (so_err == 0) {
373. connected = true;
374. } else if (so_err == 111) {
375. last = "connection refused";
376. } else {
377. last = "connection failed";
378. }
379. } else if (pr == 0) {
380. _ = c.close(fd);
381. err_out.* = "timed out";
382. return null;
383. } else {
384. last = "connection failed";
385. }
386. }
388. if (!connected) {
389. _ = c.close(fd);
390. continue;
391. }
393. _ = linux.fcntl(fd, linux.F.SETFL, flags);
394. const left = @max(deadline - nowMs(), 1);
395. var tv = c.struct_timeval{
396. .tv_sec = @intCast(@divTrunc(left, 1000)),
397. .tv_usec = @intCast(@rem(left, 1000) * 1000),
398. };
399. _ = c.setsockopt(fd, c.SOL_SOCKET, c.SO_RCVTIMEO, &tv, @sizeOf(c.struct_timeval));
400. _ = c.setsockopt(fd, c.SOL_SOCKET, c.SO_SNDTIMEO, &tv, @sizeOf(c.struct_timeval));
401. const one: c_int = 1;
402. _ = c.setsockopt(fd, c.IPPROTO_TCP, c.TCP_NODELAY, &one, @sizeOf(c_int));
403. return fd;
404. }
406. err_out.* = last;
407. return null;
408. }
410. /// The one place either transport is read or written, so the HTTP exchange below
411. /// is written once for http:// and https://.
412. const Conn = struct {
413. fd: i32,
414. ssl: ?*tls.SSL = null,
415. tls_ctx: ?*const tls.TlsContext = null,
417. fn writeAll(self: *Conn, data: []const u8) bool {
418. if (self.ssl) |ssl| {
419. return self.tls_ctx.?.writeAll(ssl, data) == data.len;
420. }
421. var sent: usize = 0;
422. while (sent < data.len) {
423. const n = c.write(self.fd, data.ptr + sent, data.len - sent);
424. if (n <= 0) return false;
425. sent += @intCast(n);
426. }
427. return true;
428. }
430. /// >0 bytes, 0 = clean end of body, <0 = error (or a socket timeout, which the
431. /// caller separates from an error by looking at the clock).
432. fn read(self: *Conn, buf: []u8) isize {
433. if (self.ssl) |ssl| {
434. const n = self.tls_ctx.?.read(ssl, buf);
435. if (n > 0) return n;
436. const e = self.tls_ctx.?.getError(ssl, n);
437. if (e == tls.SSL_ERROR_ZERO_RETURN) return 0;
438. return -1;
439. }
440. return c.read(self.fd, buf.ptr, buf.len);
441. }
443. fn close(self: *Conn) void {
444. if (self.ssl) |ssl| {
445. self.tls_ctx.?.shutdownAndFree(ssl);
446. self.ssl = null;
447. }
448. _ = c.close(self.fd);
449. }
450. };
452. // ── The client TLS context, one per CA file ──────────────────────────────────
453. //
454. // An SSL_CTX is expensive (it reads the system trust store) and thread-safe, so it
455. // is built once per distinct `caFile` and shared. The map is tiny by construction:
456. // a program has a system-trust context and at most a handful of pinned fixtures.
458. const CtxEntry = struct {
459. key: []u8,
460. ctx: tls.TlsContext,
461. };
463. var ctx_mutex: PthreadMutex = libc.PTHREAD_MUTEX_INITIALIZER;
464. var ctx_list: std.ArrayListUnmanaged(CtxEntry) = .empty;
466. fn clientCtx(ca_file: ?[]const u8, err_out: *[]const u8) ?*const tls.TlsContext {
467. const key = ca_file orelse "";
468. mutexLock(&ctx_mutex);
469. defer mutexUnlock(&ctx_mutex);
471. for (ctx_list.items) |*e| {
472. if (std.mem.eql(u8, e.key, key)) return &e.ctx;
473. }
475. const ctx = tls.TlsContext.initClient(.{ .ca_file = ca_file, .tag = "fetch" }) orelse {
476. err_out.* = "https is not available (no usable OpenSSL, or the caFile could not be loaded)";
477. return null;
478. };
479. const key_copy = allocator.dupe(u8, key) catch {
480. err_out.* = "out of memory";
481. return null;
482. };
483. ctx_list.append(allocator, .{ .key = key_copy, .ctx = ctx }) catch {
484. allocator.free(key_copy);
485. err_out.* = "out of memory";
486. return null;
487. };
488. return &ctx_list.items[ctx_list.items.len - 1].ctx;
489. }
491. // ── One HTTP exchange ────────────────────────────────────────────────────────
493. const Exchange = struct {
494. status: u16 = 0,
495. headers: []Header = &.{},
496. body: []u8 = &.{},
498. fn deinit(self: *Exchange) void {
499. freeHeaders(self.headers);
500. allocator.free(self.body);
501. }
502. };
504. fn headerValue(headers: []const Header, name: []const u8) ?[]const u8 {
505. for (headers) |h| {
506. if (std.ascii.eqlIgnoreCase(h.name, name)) return h.value;
507. }
508. return null;
509. }
511. fn hasHeader(headers: []const Header, name: []const u8) bool {
512. return headerValue(headers, name) != null;
513. }
515. /// Send one request over one connection and read the whole response.
516. /// `err_out` gets a short static reason; the caller adds the URL.
517. /// The status code off a raw head, or null while unparseable.
518. fn parseStatus(head: []const u8) ?u16 {
519. var lines = std.mem.splitSequence(u8, head, "\r\n");
520. const status_line = lines.next() orelse return null;
521. var parts = std.mem.splitScalar(u8, status_line, ' ');
522. _ = parts.next(); // HTTP/1.1
523. const code = parts.next() orelse return null;
524. return std.fmt.parseInt(u16, code, 10) catch null;
525. }
527. /// Head only → an Exchange with an EMPTY body (the chunks were the body).
528. fn parseHeadOnly(head_raw: []const u8, err_out: *[]const u8) ?Exchange {
529. // reuse the full parser on a synthetic complete response: the head as-is
530. // (its transfer-encoding stripped so no body is expected) + no body.
531. var synth: std.ArrayListUnmanaged(u8) = .empty;
532. defer synth.deinit(allocator);
533. var lines = std.mem.splitSequence(u8, head_raw[0 .. head_raw.len - 4], "\r\n");
534. while (lines.next()) |line| {
535. if (line.len > 18 and std.ascii.eqlIgnoreCase(line[0..18], "transfer-encoding:")) continue;
536. if (line.len > 15 and std.ascii.eqlIgnoreCase(line[0..15], "content-length:")) continue;
537. if (!add(&synth, line)) return oom(err_out);
538. if (!add(&synth, "\r\n")) return oom(err_out);
539. }
540. if (!add(&synth, "\r\n")) return oom(err_out);
541. return parseResponse(synth.items, err_out);
542. }
544. /// STREAM the body: post every decoded piece as a chunk frame until the peer
545. /// closes (`connection: close` is on every request, so close IS the end).
546. /// Chunked transfer is decoded incrementally; anything else streams as-is.
547. fn streamBody(t: *Ticket, conn: *Conn, raw: *std.ArrayListUnmanaged(u8), he: usize, deadline: i64, err_out: *[]const u8) ?Exchange {
548. defer raw.deinit(allocator);
549. const chunked = blk: {
550. if (findHeaderIn(raw.items[0..he], "transfer-encoding")) |v| {
551. break :blk std.ascii.indexOfIgnoreCase(v, "chunked") != null;
552. }
553. break :blk false;
554. };
556. // chunked-decoder state, carried across reads
557. var pending: std.ArrayListUnmanaged(u8) = .empty;
558. defer pending.deinit(allocator);
559. var remaining: usize = 0; // bytes left of the current chunk's data
560. var done = false;
562. // seed with whatever body bytes arrived along with the head
563. if (raw.items.len > he) {
564. pending.appendSlice(allocator, raw.items[he..]) catch return oom(err_out);
565. }
566. var buf: [16 * 1024]u8 = undefined;
567. while (true) {
568. // decode + post what we hold
569. if (!chunked) {
570. if (pending.items.len > 0) {
571. postChunk(t, pending.items);
572. pending.clearRetainingCapacity();
573. }
574. } else while (!done) {
575. if (remaining > 0) {
576. const take = @min(remaining, pending.items.len);
577. if (take == 0) break;
578. postChunk(t, pending.items[0..take]);
579. std.mem.copyForwards(u8, pending.items, pending.items[take..]);
580. pending.shrinkRetainingCapacity(pending.items.len - take);
581. remaining -= take;
582. if (remaining == 0) {
583. // the CRLF after the chunk data
584. if (pending.items.len < 2) break;
585. std.mem.copyForwards(u8, pending.items, pending.items[2..]);
586. pending.shrinkRetainingCapacity(pending.items.len - 2);
587. }
588. continue;
589. }
590. const nl = std.mem.indexOf(u8, pending.items, "\r\n") orelse break;
591. const size_line = pending.items[0..nl];
592. const semi = std.mem.indexOfScalar(u8, size_line, ';') orelse size_line.len;
593. const n = std.fmt.parseInt(usize, std.mem.trim(u8, size_line[0..semi], " \t"), 16) catch {
594. err_out.* = "the peer sent a malformed chunked body";
595. return null;
596. };
597. std.mem.copyForwards(u8, pending.items, pending.items[nl + 2 ..]);
598. pending.shrinkRetainingCapacity(pending.items.len - nl - 2);
599. if (n == 0) {
600. done = true;
601. break;
602. }
603. remaining = n;
604. }
605. if (done) break;
607. const n = conn.read(&buf);
608. if (n == 0) break; // close IS the end (identity), or a truncated chunked stream
609. if (n < 0) {
610. if (nowMs() >= deadline) {
611. err_out.* = "timed out";
612. return null;
613. }
614. break;
615. }
616. pending.appendSlice(allocator, buf[0..@intCast(n)]) catch return oom(err_out);
617. }
619. return parseHeadOnly(raw.items[0..he], err_out);
620. }
622. fn exchange(t: *Ticket, url: Url, method: []const u8, body: []const u8, send_body: bool, deadline: i64, err_out: *[]const u8) ?Exchange {
623. const budget = deadline - nowMs();
624. const fd = tcpConnect(url.host, url.port, budget, err_out) orelse return null;
625. if (!t.setLiveFd(fd)) {
626. _ = c.close(fd);
627. err_out.* = "aborted";
628. return null;
629. }
631. var conn = Conn{ .fd = fd };
632. if (url.https) {
633. const ctx = clientCtx(t.req.ca_file, err_out) orelse {
634. _ = c.close(fd);
635. return null;
636. };
637. // 137's shared client half logs the specific failure itself; the caller
638. // gets the located summary (certificate not trusted, or not a TLS server)
639. const ssl = ctx.connect(fd, url.host, "fetch") orelse {
640. _ = c.close(fd);
641. err_out.* = "TLS handshake failed (certificate not trusted, or not a TLS server)";
642. return null;
643. };
644. conn.ssl = ssl;
645. conn.tls_ctx = ctx;
646. }
647. defer conn.close();
648. defer t.clearLiveFd();
650. // --- request ---
651. var req_buf: std.ArrayListUnmanaged(u8) = .empty;
652. defer req_buf.deinit(allocator);
654. const default_port: u16 = if (url.https) 443 else 80;
655. var ok = true;
656. ok = ok and addFmt(&req_buf, "{s} {s} HTTP/1.1\r\n", .{ method, url.target });
657. if (url.port == default_port) {
658. ok = ok and addFmt(&req_buf, "host: {s}\r\n", .{url.host});
659. } else {
660. ok = ok and addFmt(&req_buf, "host: {s}:{d}\r\n", .{ url.host, url.port });
661. }
662. // `connection: close` on every request: there is no connection pool, and a
663. // closed connection is also the unambiguous end of a body with no length.
664. ok = ok and add(&req_buf, "connection: close\r\n");
665. if (!hasHeader(t.req.headers, "user-agent")) {
666. ok = ok and add(&req_buf, "user-agent: hybriel-fetch/1\r\n");
667. }
668. if (!hasHeader(t.req.headers, "accept")) {
669. ok = ok and add(&req_buf, "accept: */*\r\n");
670. }
671. if (send_body) {
672. ok = ok and addFmt(&req_buf, "content-length: {d}\r\n", .{body.len});
673. if (t.req.json_body and !hasHeader(t.req.headers, "content-type")) {
674. ok = ok and add(&req_buf, "content-type: application/json\r\n");
675. }
676. }
677. for (t.req.headers) |h| {
678. // A caller header may not smuggle a second request in (CRLF injection).
679. if (std.mem.indexOfAny(u8, h.name, "\r\n") != null or std.mem.indexOfAny(u8, h.value, "\r\n") != null) {
680. err_out.* = "a header name or value contains a line break";
681. return null;
682. }
683. ok = ok and addFmt(&req_buf, "{s}: {s}\r\n", .{ h.name, h.value });
684. }
685. ok = ok and add(&req_buf, "\r\n");
686. if (send_body and body.len > 0) {
687. ok = ok and add(&req_buf, body);
688. }
689. if (!ok) return oom(err_out);
691. if (!conn.writeAll(req_buf.items)) {
692. err_out.* = if (nowMs() >= deadline) "timed out" else "could not send the request";
693. return null;
694. }
696. // --- response ---
697. var raw: std.ArrayListUnmanaged(u8) = .empty;
698. errdefer raw.deinit(allocator);
699. var head_end: ?usize = null;
700. var buf: [16 * 1024]u8 = undefined;
702. while (true) {
703. if (head_end == null) {
704. if (std.mem.indexOf(u8, raw.items, "\r\n\r\n")) |idx| head_end = idx + 4;
705. }
706. if (head_end) |he| {
707. // STREAMING (LLM tokens): the head is in — if this is the final
708. // answer (not a redirect the caller follows), hand the rest of the
709. // body over CHUNK BY CHUNK as it arrives instead of accumulating.
710. if (t.req.stream) {
711. const status_ok = parseStatus(raw.items[0..he]);
712. if (status_ok != null and !isRedirect(status_ok.?)) {
713. return streamBody(t, &conn, &raw, he, deadline, err_out);
714. }
715. }
716. // Once the head is in, stop as soon as the body is provably complete.
717. if (bodyComplete(raw.items, he)) break;
718. }
719. if (raw.items.len > MAX_BODY) {
720. raw.deinit(allocator);
721. err_out.* = "the response is larger than the 32 MiB limit";
722. return null;
723. }
724. const n = conn.read(&buf);
725. if (n == 0) break;
726. if (n < 0) {
727. if (nowMs() >= deadline) {
728. raw.deinit(allocator);
729. err_out.* = "timed out";
730. return null;
731. }
732. // A TLS peer that just closes without close_notify, or a plain socket
733. // reset after a complete answer: the parse below decides whether what
734. // arrived is a whole response.
735. break;
736. }
737. raw.appendSlice(allocator, buf[0..@intCast(n)]) catch {
738. raw.deinit(allocator);
739. return oom(err_out);
740. };
741. }
743. const out = parseResponse(raw.items, err_out) orelse {
744. raw.deinit(allocator);
745. return null;
746. };
747. raw.deinit(allocator);
748. return out;
749. }
751. fn add(buf: *std.ArrayListUnmanaged(u8), data: []const u8) bool {
752. buf.appendSlice(allocator, data) catch return false;
753. return true;
754. }
756. /// One request line or header, formatted. A header that does not fit 2 KiB is not
757. /// a header anyone meant to send.
758. fn addFmt(buf: *std.ArrayListUnmanaged(u8), comptime fmt: []const u8, args: anytype) bool {
759. var tmp: [2048]u8 = undefined;
760. const s = std.fmt.bufPrint(&tmp, fmt, args) catch return false;
761. return add(buf, s);
762. }
764. fn oom(err_out: *[]const u8) ?Exchange {
765. err_out.* = "out of memory";
766. return null;
767. }
769. /// Is everything the response promised already in `raw`? Content-Length is exact,
770. /// chunked ends at the zero chunk, and anything else ends when the peer closes.
771. fn bodyComplete(raw: []const u8, head_end: usize) bool {
772. const head = raw[0..head_end];
773. if (findHeaderIn(head, "content-length")) |v| {
774. const want = std.fmt.parseInt(usize, std.mem.trim(u8, v, " \t"), 10) catch return false;
775. return raw.len - head_end >= want;
776. }
777. if (findHeaderIn(head, "transfer-encoding")) |v| {
778. if (std.ascii.indexOfIgnoreCase(v, "chunked") != null) {
779. return std.mem.indexOf(u8, raw[head_end..], "\r\n0\r\n") != null or
780. std.mem.startsWith(u8, raw[head_end..], "0\r\n");
781. }
782. }
783. return false;
784. }
786. fn findHeaderIn(head: []const u8, name: []const u8) ?[]const u8 {
787. var lines = std.mem.splitSequence(u8, head, "\r\n");
788. _ = lines.next(); // status line
789. while (lines.next()) |line| {
790. if (line.len == 0) break;
791. const colon = std.mem.indexOfScalar(u8, line, ':') orelse continue;
792. if (std.ascii.eqlIgnoreCase(std.mem.trim(u8, line[0..colon], " \t"), name)) {
793. return std.mem.trim(u8, line[colon + 1 ..], " \t");
794. }
795. }
796. return null;
797. }
799. fn parseResponse(raw: []const u8, err_out: *[]const u8) ?Exchange {
800. const he = std.mem.indexOf(u8, raw, "\r\n\r\n") orelse {
801. err_out.* = "the peer sent no complete HTTP response";
802. return null;
803. };
804. const head = raw[0..he];
805. const body_raw = raw[he + 4 ..];
807. var lines = std.mem.splitSequence(u8, head, "\r\n");
808. const status_line = lines.next() orelse {
809. err_out.* = "the peer sent no status line";
810. return null;
811. };
812. // "HTTP/1.1 200 OK"
813. const sp = std.mem.indexOfScalar(u8, status_line, ' ') orelse {
814. err_out.* = "the peer sent a malformed status line";
815. return null;
816. };
817. const after = status_line[sp + 1 ..];
818. const code_end = std.mem.indexOfScalar(u8, after, ' ') orelse after.len;
819. const status = std.fmt.parseInt(u16, std.mem.trim(u8, after[0..code_end], " \t"), 10) catch {
820. err_out.* = "the peer sent a malformed status line";
821. return null;
822. };
824. var headers: std.ArrayListUnmanaged(Header) = .empty;
825. errdefer {
826. for (headers.items) |h| {
827. allocator.free(h.name);
828. allocator.free(h.value);
829. }
830. headers.deinit(allocator);
831. }
832. var chunked = false;
833. while (lines.next()) |line| {
834. if (line.len == 0) continue;
835. const colon = std.mem.indexOfScalar(u8, line, ':') orelse continue;
836. const name = std.mem.trim(u8, line[0..colon], " \t");
837. const value = std.mem.trim(u8, line[colon + 1 ..], " \t");
838. if (name.len == 0) continue;
839. const name_copy = allocator.dupe(u8, name) catch return oom(err_out);
840. http.toLowercase(name_copy);
841. const value_copy = allocator.dupe(u8, value) catch {
842. allocator.free(name_copy);
843. return oom(err_out);
844. };
845. if (std.mem.eql(u8, name_copy, "transfer-encoding") and
846. std.ascii.indexOfIgnoreCase(value_copy, "chunked") != null) chunked = true;
847. headers.append(allocator, .{ .name = name_copy, .value = value_copy }) catch {
848. allocator.free(name_copy);
849. allocator.free(value_copy);
850. return oom(err_out);
851. };
852. }
854. var body: []u8 = undefined;
855. if (chunked) {
856. body = dechunk(body_raw) orelse {
857. for (headers.items) |h| {
858. allocator.free(h.name);
859. allocator.free(h.value);
860. }
861. headers.deinit(allocator);
862. err_out.* = "the peer sent a malformed chunked body";
863. return null;
864. };
865. } else if (findHeaderIn(head, "content-length")) |v| {
866. const want = std.fmt.parseInt(usize, v, 10) catch body_raw.len;
867. const take = @min(want, body_raw.len);
868. body = allocator.dupe(u8, body_raw[0..take]) catch return oom(err_out);
869. } else {
870. body = allocator.dupe(u8, body_raw) catch return oom(err_out);
871. }
873. const headers_slice = headers.toOwnedSlice(allocator) catch return oom(err_out);
874. return .{ .status = status, .headers = headers_slice, .body = body };
875. }
877. fn dechunk(input: []const u8) ?[]u8 {
878. var out: std.ArrayListUnmanaged(u8) = .empty;
879. var i: usize = 0;
880. while (true) {
881. const line_end = std.mem.indexOfPos(u8, input, i, "\r\n") orelse {
882. out.deinit(allocator);
883. return null;
884. };
885. var size_txt = input[i..line_end];
886. if (std.mem.indexOfScalar(u8, size_txt, ';')) |semi| size_txt = size_txt[0..semi];
887. const size = std.fmt.parseInt(usize, std.mem.trim(u8, size_txt, " \t"), 16) catch {
888. out.deinit(allocator);
889. return null;
890. };
891. i = line_end + 2;
892. if (size == 0) return out.toOwnedSlice(allocator) catch null;
893. if (i + size > input.len) {
894. out.deinit(allocator);
895. return null;
896. }
897. out.appendSlice(allocator, input[i .. i + size]) catch {
898. out.deinit(allocator);
899. return null;
900. };
901. i += size + 2; // skip the chunk's trailing CRLF
902. if (i > input.len) return out.toOwnedSlice(allocator) catch null;
903. }
904. }
906. // ── The worker: one thread per fetch, redirects and all ───────────────────────
908. fn isRedirect(status: u16) bool {
909. return status == 301 or status == 302 or status == 303 or status == 307 or status == 308;
910. }
913. // ── The ACTIVE registry: every in-flight ticket, so `fetchAbort(tag)` can find
914. // its socket. Added at spawn, removed at finish — a finished ticket is owned by
915. // its handle alone and an abort of it is a no-op.
916. var active_mutex: PthreadMutex = libc.PTHREAD_MUTEX_INITIALIZER;
917. var active_tickets: std.ArrayListUnmanaged(*Ticket) = .empty;
919. fn activeAdd(t: *Ticket) void {
920. mutexLock(&active_mutex);
921. active_tickets.append(allocator, t) catch {};
922. mutexUnlock(&active_mutex);
923. }
925. fn activeRemove(t: *Ticket) void {
926. mutexLock(&active_mutex);
927. for (active_tickets.items, 0..) |a, i| {
928. if (a == t) {
929. _ = active_tickets.swapRemove(i);
930. break;
931. }
932. }
933. mutexUnlock(&active_mutex);
934. }
936. fn workerMain(t: *Ticket) void {
937. runFetch(t);
938. activeRemove(t);
939. // an aborted request reports ABORTED whatever the read error looked like —
940. // the app asked for the interrupt, the message says the interrupt happened
941. if (t.aborted) {
942. if (t.err) |e| allocator.free(e);
943. t.err = std.fmt.allocPrint(allocator, "hl:fetch: aborted", .{}) catch null;
944. }
945. t.finish();
946. postCompletion(t);
947. }
951. fn runFetch(t: *Ticket) void {
952. const deadline = nowMs() + t.req.timeout_ms;
954. var current = allocator.dupe(u8, t.req.url) catch {
955. t.fail("hl:fetch: out of memory", .{});
956. return;
957. };
958. var method = allocator.dupe(u8, t.req.method) catch {
959. allocator.free(current);
960. t.fail("hl:fetch: out of memory", .{});
961. return;
962. };
963. var send_body = t.req.has_body;
964. var hops: u32 = 0;
966. while (true) {
967. const url = parseUrl(current) orelse {
968. t.fail("hl:fetch: '{s}' is not an http:// or https:// URL", .{current});
969. allocator.free(current);
970. allocator.free(method);
971. return;
972. };
974. var reason: []const u8 = "connection failed";
975. var ex = exchange(t, url, method, t.req.body, send_body, deadline, &reason) orelse {
976. t.fail("hl:fetch {s} {s}: {s}", .{ method, current, reason });
977. allocator.free(current);
978. allocator.free(method);
979. return;
980. };
982. if (isRedirect(ex.status) and hops < t.req.max_redirects) {
983. if (headerValue(ex.headers, "location")) |loc| {
984. const next = resolveLocation(current, loc) orelse {
985. t.fail("hl:fetch {s}: redirect to an unusable Location '{s}'", .{ current, loc });
986. ex.deinit();
987. allocator.free(current);
988. allocator.free(method);
989. return;
990. };
991. // 303 always becomes a GET; 301/302 become one for anything that
992. // is not already GET/HEAD — what every browser does, and what a
993. // server that answers a POST with "see the result over there"
994. // means. 307/308 keep the method AND the body by definition.
995. if (ex.status == 303 or
996. ((ex.status == 301 or ex.status == 302) and
997. !std.mem.eql(u8, method, "GET") and !std.mem.eql(u8, method, "HEAD")))
998. {
999. allocator.free(method);
1000. method = allocator.dupe(u8, "GET") catch {
1001. allocator.free(next);
1002. ex.deinit();
1003. allocator.free(current);
1004. t.fail("hl:fetch: out of memory", .{});
1005. return;
1006. };
1007. send_body = false;
1008. }
1009. ex.deinit();
1010. allocator.free(current);
1011. current = next;
1012. hops += 1;
1013. continue;
1014. }
1015. }
1017. if (isRedirect(ex.status) and t.req.max_redirects > 0 and hops >= t.req.max_redirects) {
1018. t.fail("hl:fetch {s}: more than {d} redirects", .{ t.req.url, t.req.max_redirects });
1019. ex.deinit();
1020. allocator.free(current);
1021. allocator.free(method);
1022. return;
1023. }
1025. t.status = ex.status;
1026. t.headers = ex.headers;
1027. t.body = ex.body;
1028. t.final_url = current;
1029. allocator.free(method);
1030. return;
1031. }
1032. }
1034. // ── The value shapes handed back to Hybriel ──────────────────────────────────
1036. fn hlStr(s: []const u8) api.HlString {
1037. return .{ .ptr = s.ptr, .len = s.len };
1038. }
1040. /// Only the SHELLS are freed: every string in a `result()` object points into the
1041. /// ticket, which outlives the call (the loader copies what it keeps before this
1042. /// runs — the plugin ABI's ownership rule).
1043. fn respShellDeinit(obj: *HlObject) callconv(.c) void {
1044. const hdrs = obj.fields[3].value;
1045. if (hdrs.type == .hl_object) {
1046. const inner = hdrs.data.object;
1047. allocator.free(inner.fields[0..inner.field_count]);
1048. allocator.destroy(inner);
1049. }
1050. allocator.free(obj.fields[0..obj.field_count]);
1051. allocator.destroy(obj);
1052. }
1054. fn buildHeadersObject(headers: []const Header) ?*HlObject {
1055. const fields = allocator.alloc(HlField, headers.len) catch return null;
1056. for (headers, 0..) |h, i| {
1057. fields[i] = .{ .key = hlStr(h.name), .value = api.makeString(h.value) };
1058. }
1059. const obj = allocator.create(HlObject) catch {
1060. allocator.free(fields);
1061. return null;
1062. };
1063. obj.* = .{ .fields = fields.ptr, .field_count = headers.len, .deinit_fn = null };
1064. return obj;
1065. }
1067. /// `{ status, ok, url, headers, body, tag }` — the response as the .hl side sees
1068. /// it, hydrated into a FetchResponse by server.hl.
1069. fn buildResponse(t: *Ticket) HlValue {
1070. const headers_obj = buildHeadersObject(t.headers) orelse return api.makeError("hl:fetch: out of memory");
1071. const fields = allocator.alloc(HlField, 6) catch {
1072. allocator.free(headers_obj.fields[0..headers_obj.field_count]);
1073. allocator.destroy(headers_obj);
1074. return api.makeError("hl:fetch: out of memory");
1075. };
1076. fields[0] = .{ .key = hlStr("status"), .value = api.makeNumber(@floatFromInt(t.status)) };
1077. fields[1] = .{ .key = hlStr("ok"), .value = api.makeBool(t.status >= 200 and t.status < 300) };
1078. fields[2] = .{ .key = hlStr("url"), .value = api.makeString(t.final_url) };
1079. fields[3] = .{ .key = hlStr("headers"), .value = api.makeObject(headers_obj) };
1080. fields[4] = .{ .key = hlStr("body"), .value = api.makeString(t.body) };
1081. fields[5] = .{ .key = hlStr("tag"), .value = api.makeString(t.req.tag) };
1083. const obj = allocator.create(HlObject) catch {
1084. allocator.free(fields);
1085. allocator.free(headers_obj.fields[0..headers_obj.field_count]);
1086. allocator.destroy(headers_obj);
1087. return api.makeError("hl:fetch: out of memory");
1088. };
1089. obj.* = .{ .fields = fields.ptr, .field_count = 6, .deinit_fn = &respShellDeinit };
1090. return api.makeObject(obj);
1091. }
1093. // ── The ticket handle ────────────────────────────────────────────────────────
1095. fn ticketCall(ctx: ?*anyopaque, op: api.HlString, argc: u32, argv: [*]const HlValue) callconv(.c) HlValue {
1096. _ = argc;
1097. _ = argv;
1098. const t: *Ticket = @ptrCast(@alignCast(ctx orelse return api.makeError("hl:fetch: dead ticket")));
1099. const name = op.ptr[0..op.len];
1101. if (std.mem.eql(u8, name, "done")) {
1102. return api.makeBool(t.isDone());
1103. }
1104. if (std.mem.eql(u8, name, "result")) {
1105. t.wait();
1106. if (t.err) |msg| return api.makeError(msg);
1107. return buildResponse(t);
1108. }
1109. if (std.mem.eql(u8, name, "events")) {
1110. if (t.ev_wake_fd < 0) return api.makeError("hl:fetch: events() is stream mode — start the fetch with stream = true");
1111. const iter = allocator.create(HlIterator) catch return api.makeError("hl:fetch: out of memory");
1112. iter.* = .{
1113. .context = @ptrCast(t),
1114. .next_fn = &eventsTryNext, // non-blocking either way — event loop only
1115. .deinit_fn = null,
1116. .try_next_fn = &eventsTryNext,
1117. .wake_fd = t.ev_wake_fd,
1118. };
1119. return api.makeIterator(iter);
1120. }
1121. if (std.mem.eql(u8, name, "abort")) {
1122. t.abort();
1123. return api.makeBool(true);
1124. }
1125. return api.makeError("hl:fetch: a pending fetch has no method by that name");
1126. }
1128. fn ticketClose(ctx: ?*anyopaque) callconv(.c) void {
1129. const t: *Ticket = @ptrCast(@alignCast(ctx orelse return));
1130. // The worker writes into this ticket, so it is JOINED before anything is
1131. // freed — a detached thread finishing after the free is a use-after-free the
1132. // collector would surface as random corruption. A dropped, never-read fetch
1133. // therefore costs at most its own timeout at collection time.
1134. if (t.thread) |th| {
1135. th.join();
1136. t.thread = null;
1137. }
1138. t.req.deinit();
1139. for (t.ev_queue.items) |comp| {
1140. allocator.free(comp.tag);
1141. allocator.free(comp.url);
1142. allocator.free(comp.body);
1143. allocator.free(comp.err);
1144. if (comp.chunk) |ch| allocator.free(ch);
1145. for (comp.headers) |h| {
1146. allocator.free(h.name);
1147. allocator.free(h.value);
1148. }
1149. if (comp.headers.len > 0) allocator.free(comp.headers);
1150. }
1151. t.ev_queue.deinit(allocator);
1152. freeHeaders(t.headers);
1153. allocator.free(t.body);
1154. allocator.free(t.final_url);
1155. if (t.err) |e| allocator.free(e);
1156. allocator.destroy(t);
1157. }
1159. // ── Argument reading ─────────────────────────────────────────────────────────
1161. fn fieldOf(v: HlValue, name: []const u8) ?HlValue {
1162. if (v.type != .hl_object) return null;
1163. const obj = v.data.object;
1164. for (obj.fields[0..obj.field_count]) |f| {
1165. if (std.mem.eql(u8, f.key.ptr[0..f.key.len], name)) return f.value;
1166. }
1167. return null;
1168. }
1170. fn dupString(v: ?HlValue, fallback: []const u8) []u8 {
1171. if (v) |val| {
1172. if (val.type == .hl_string) {
1173. return allocator.dupe(u8, val.data.string.ptr[0..val.data.string.len]) catch
1174. allocator.dupe(u8, fallback) catch unreachable;
1175. }
1176. }
1177. return allocator.dupe(u8, fallback) catch unreachable;
1178. }
1180. fn numberOr(v: ?HlValue, fallback: i64) i64 {
1181. if (v) |val| {
1182. if (val.type == .hl_number) return @intFromFloat(val.data.number);
1183. }
1184. return fallback;
1185. }
1187. fn boolOr(v: ?HlValue, fallback: bool) bool {
1188. if (v) |val| {
1189. if (val.type == .hl_bool) return val.data.boolean;
1190. }
1191. return fallback;
1192. }
1194. fn readHeaders(v: ?HlValue) []Header {
1195. const val = v orelse return &.{};
1196. if (val.type != .hl_object) return &.{};
1197. const obj = val.data.object;
1198. var list: std.ArrayListUnmanaged(Header) = .empty;
1199. for (obj.fields[0..obj.field_count]) |f| {
1200. if (f.value.type != .hl_string) continue;
1201. const name = allocator.dupe(u8, f.key.ptr[0..f.key.len]) catch continue;
1202. const value = allocator.dupe(u8, f.value.data.string.ptr[0..f.value.data.string.len]) catch {
1203. allocator.free(name);
1204. continue;
1205. };
1206. list.append(allocator, .{ .name = name, .value = value }) catch {
1207. allocator.free(name);
1208. allocator.free(value);
1209. };
1210. }
1211. return list.toOwnedSlice(allocator) catch &.{};
1212. }
1214. /// __native("fetch.start", url, options) → a ticket HANDLE, immediately.
1215. /// Options: method, headers, body, jsonBody, timeoutMs, maxRedirects, caFile, tag.
1216. export fn hl_fetch_start(argc: u32, argv: [*]const HlValue) callconv(.c) HlValue {
1217. if (argc < 1 or argv[0].type != .hl_string) {
1218. return api.makeError("hl:fetch: fetch() needs a URL string");
1219. }
1220. const url = argv[0].data.string.ptr[0..argv[0].data.string.len];
1221. const opts: ?HlValue = if (argc > 1 and argv[1].type == .hl_object) argv[1] else null;
1223. const method_raw = dupString(if (opts) |o| fieldOf(o, "method") else null, "GET");
1224. for (method_raw) |*ch| {
1225. if (ch.* >= 'a' and ch.* <= 'z') ch.* -= 32;
1226. }
1227. const body = dupString(if (opts) |o| fieldOf(o, "body") else null, "");
1228. const has_body = blk: {
1229. const bv = if (opts) |o| fieldOf(o, "body") else null;
1230. if (bv) |v| break :blk v.type == .hl_string;
1231. break :blk false;
1232. };
1233. const ca_val = if (opts) |o| fieldOf(o, "caFile") else null;
1234. const ca_file: ?[]u8 = if (ca_val != null and ca_val.?.type == .hl_string)
1235. allocator.dupe(u8, ca_val.?.data.string.ptr[0..ca_val.?.data.string.len]) catch null
1236. else
1237. null;
1239. const t = allocator.create(Ticket) catch return api.makeError("hl:fetch: out of memory");
1240. t.* = .{
1241. .req = .{
1242. .method = method_raw,
1243. .url = allocator.dupe(u8, url) catch return api.makeError("hl:fetch: out of memory"),
1244. .headers = readHeaders(if (opts) |o| fieldOf(o, "headers") else null),
1245. .body = body,
1246. .has_body = has_body,
1247. .json_body = boolOr(if (opts) |o| fieldOf(o, "jsonBody") else null, false),
1248. .timeout_ms = @max(numberOr(if (opts) |o| fieldOf(o, "timeoutMs") else null, DEFAULT_TIMEOUT_MS), 1),
1249. .max_redirects = @intCast(@max(numberOr(if (opts) |o| fieldOf(o, "maxRedirects") else null, DEFAULT_MAX_REDIRECTS), 0)),
1250. .ca_file = ca_file,
1251. .tag = dupString(if (opts) |o| fieldOf(o, "tag") else null, ""),
1252. .stream = boolOr(if (opts) |o| fieldOf(o, "stream") else null, false),
1253. },
1254. };
1256. if (t.req.stream) t.ev_wake_fd = http.makeWakeFd();
1258. activeAdd(t);
1259. t.thread = std.Thread.spawn(.{}, workerMain, .{t}) catch {
1260. // No thread: run it here rather than lying about having started. The call
1261. // then blocks, which is slower and still correct.
1262. workerMain(t);
1263. t.thread = null;
1264. return makeTicketHandle(t);
1265. };
1266. return makeTicketHandle(t);
1267. }
1269. fn makeTicketHandle(t: *Ticket) HlValue {
1270. const h = allocator.create(HlHandle) catch return api.makeError("hl:fetch: out of memory");
1271. h.* = .{
1272. .context = @ptrCast(t),
1273. .call_fn = &ticketCall,
1274. .close_fn = &ticketClose,
1275. .type_name = hlStr("pending fetch"),
1276. };
1277. return api.makeHandle(h);
1278. }
1280. // ── The completions source: a loop source with the mission-125 bell ──────────
1281. //
1282. // `for (ev of completions())` / `eventloop.register` — the SAME shape hl:http1's
1283. // request and ws-event sources have, so a finished fetch reaches an `on fetched(ev)`
1284. // handler through the loop's own `epoll_wait` instead of anyone polling. The queue
1285. // is filled by the worker threads and drained by the loop thread.
1287. const Completion = struct {
1288. aborted: bool = false,
1289. tag: []u8,
1290. url: []u8,
1291. status: u16,
1292. headers: []Header,
1293. body: []u8,
1294. err: []u8,
1295. /// non-null = a STREAM CHUNK frame ({tag, url, chunk}), not a completion
1296. chunk: ?[]u8 = null,
1297. };
1299. var comp_mutex: PthreadMutex = libc.PTHREAD_MUTEX_INITIALIZER;
1300. var comp_queue: std.ArrayListUnmanaged(Completion) = .empty;
1301. var comp_wake_fd: i32 = -1;
1302. var comp_enabled = std.atomic.Value(bool).init(false);
1304. /// Called by every worker as its last act. Nothing is copied — and no wake is
1305. /// rung — unless a completions source actually exists, so a program that only
1306. /// uses `result()` pays nothing for this.
1307. /// One decoded piece of a STREAMING body → a `{ tag, url, chunk }` frame on the
1308. /// completions source, bell rung. Dropped (with one log line) when no source is
1309. /// armed — a stream nobody listens to is a programming error worth seeing.
1310. fn postChunk(t: *Ticket, bytes: []const u8) void {
1311. if (t.ev_wake_fd < 0 and !comp_enabled.load(.acquire)) {
1312. logMsg("hl:fetch: stream chunk dropped — nobody is listening");
1313. return;
1314. }
1315. const comp = Completion{
1316. .tag = allocator.dupe(u8, t.req.tag) catch return,
1317. .url = allocator.dupe(u8, t.req.url) catch return,
1318. .status = 0,
1319. .headers = &.{},
1320. .body = allocator.dupe(u8, "") catch return,
1321. .err = allocator.dupe(u8, "") catch return,
1322. .chunk = allocator.dupe(u8, bytes) catch return,
1323. };
1324. postFrame(t, comp);
1325. }
1327. /// One frame to wherever this ticket's listener lives: the ticket's own source
1328. /// (stream mode) or the global completions source.
1329. fn postFrame(t: *Ticket, comp: Completion) void {
1330. if (t.ev_wake_fd >= 0) {
1331. mutexLock(&t.mutex);
1332. t.ev_queue.append(allocator, comp) catch {};
1333. mutexUnlock(&t.mutex);
1334. http.ringWake(t.ev_wake_fd);
1335. return;
1336. }
1337. mutexLock(&comp_mutex);
1338. comp_queue.append(allocator, comp) catch {};
1339. mutexUnlock(&comp_mutex);
1340. http.ringWake(comp_wake_fd);
1341. }
1343. fn postCompletion(t: *Ticket) void {
1344. if (t.ev_wake_fd < 0 and !comp_enabled.load(.acquire)) return;
1346. var headers = allocator.alloc(Header, t.headers.len) catch return;
1347. var built: usize = 0;
1348. for (t.headers, 0..) |h, i| {
1349. const n = allocator.dupe(u8, h.name) catch break;
1350. const v = allocator.dupe(u8, h.value) catch {
1351. allocator.free(n);
1352. break;
1353. };
1354. headers[i] = .{ .name = n, .value = v };
1355. built = i + 1;
1356. }
1357. headers = headers[0..built];
1359. const comp = Completion{
1360. .aborted = t.aborted,
1361. .tag = allocator.dupe(u8, t.req.tag) catch "",
1362. .url = allocator.dupe(u8, if (t.final_url.len > 0) t.final_url else t.req.url) catch "",
1363. .status = t.status,
1364. .headers = headers,
1365. .body = allocator.dupe(u8, t.body) catch "",
1366. .err = allocator.dupe(u8, if (t.err) |e| e else "") catch "",
1367. };
1369. // Ring generously (D-125): the loop treats the fd as a bell, so a wake for a
1370. // queue somebody already drained costs one empty round and nothing else.
1371. postFrame(t, comp);
1372. }
1374. /// Frees the SHELLS and the payload: a completion is a snapshot nobody else owns.
1375. fn compObjDeinit(obj: *HlObject) callconv(.c) void {
1376. const fields = obj.fields[0..obj.field_count];
1377. for (fields) |f| {
1378. if (f.value.type == .hl_string) {
1379. const s = f.value.data.string;
1380. if (s.len > 0) allocator.free(@constCast(s.ptr[0..s.len]));
1381. } else if (f.value.type == .hl_object) {
1382. const inner = f.value.data.object;
1383. for (inner.fields[0..inner.field_count]) |hf| {
1384. allocator.free(@constCast(hf.key.ptr[0..hf.key.len]));
1385. const hv = hf.value.data.string;
1386. allocator.free(@constCast(hv.ptr[0..hv.len]));
1387. }
1388. allocator.free(inner.fields[0..inner.field_count]);
1389. allocator.destroy(inner);
1390. }
1391. }
1392. allocator.free(fields);
1393. allocator.destroy(obj);
1394. }
1396. fn completionsTryNext(ctx: ?*anyopaque) callconv(.c) HlValue {
1397. _ = ctx;
1398. mutexLock(&comp_mutex);
1399. if (comp_queue.items.len == 0) {
1400. mutexUnlock(&comp_mutex);
1401. return api.makeNull();
1402. }
1403. const comp = comp_queue.orderedRemove(0);
1404. mutexUnlock(&comp_mutex);
1405. return frameToValue(comp);
1406. }
1408. /// A streaming ticket's own source: drains that ticket's queue and nothing else.
1409. fn eventsTryNext(ctx: ?*anyopaque) callconv(.c) HlValue {
1410. const t: *Ticket = @ptrCast(@alignCast(ctx orelse return api.makeNull()));
1411. mutexLock(&t.mutex);
1412. if (t.ev_queue.items.len == 0) {
1413. mutexUnlock(&t.mutex);
1414. return api.makeNull();
1415. }
1416. const comp = t.ev_queue.orderedRemove(0);
1417. mutexUnlock(&t.mutex);
1418. return frameToValue(comp);
1419. }
1421. fn frameToValue(comp: Completion) HlValue {
1422. // a STREAM CHUNK frame: { tag, url, chunk } and nothing else
1423. if (comp.chunk) |chunk| {
1424. allocator.free(comp.body);
1425. allocator.free(comp.err);
1426. const cfields = allocator.alloc(HlField, 3) catch return api.makeNull();
1427. cfields[0] = .{ .key = hlStr("tag"), .value = api.makeString(comp.tag) };
1428. cfields[1] = .{ .key = hlStr("url"), .value = api.makeString(comp.url) };
1429. cfields[2] = .{ .key = hlStr("chunk"), .value = api.makeString(chunk) };
1430. const cobj = allocator.create(HlObject) catch {
1431. allocator.free(cfields);
1432. return api.makeNull();
1433. };
1434. cobj.* = .{ .fields = cfields.ptr, .field_count = 3, .deinit_fn = &compObjDeinit };
1435. return api.makeObject(cobj);
1436. }
1438. const hdr_fields = allocator.alloc(HlField, comp.headers.len) catch return api.makeNull();
1439. for (comp.headers, 0..) |h, i| {
1440. hdr_fields[i] = .{ .key = hlStr(h.name), .value = api.makeString(h.value) };
1441. }
1442. const hdr_obj = allocator.create(HlObject) catch {
1443. allocator.free(hdr_fields);
1444. return api.makeNull();
1445. };
1446. hdr_obj.* = .{ .fields = hdr_fields.ptr, .field_count = comp.headers.len, .deinit_fn = null };
1447. allocator.free(comp.headers);
1449. const fields = allocator.alloc(HlField, 8) catch return api.makeNull();
1450. fields[0] = .{ .key = hlStr("tag"), .value = api.makeString(comp.tag) };
1451. fields[1] = .{ .key = hlStr("url"), .value = api.makeString(comp.url) };
1452. fields[2] = .{ .key = hlStr("status"), .value = api.makeNumber(@floatFromInt(comp.status)) };
1453. fields[3] = .{ .key = hlStr("ok"), .value = api.makeBool(comp.status >= 200 and comp.status < 300) };
1454. fields[4] = .{ .key = hlStr("body"), .value = api.makeString(comp.body) };
1455. fields[5] = .{ .key = hlStr("error"), .value = api.makeString(comp.err) };
1456. fields[6] = .{ .key = hlStr("headers"), .value = api.makeObject(hdr_obj) };
1457. fields[7] = .{ .key = hlStr("aborted"), .value = api.makeBool(comp.aborted) };
1459. const obj = allocator.create(HlObject) catch {
1460. allocator.free(fields);
1461. return api.makeNull();
1462. };
1463. obj.* = .{ .fields = fields.ptr, .field_count = 8, .deinit_fn = &compObjDeinit };
1464. return api.makeObject(obj);
1465. }
1467. /// __native("fetch.abort", tag) — THE INTERRUPT (the LLM stop button): every
1468. /// in-flight fetch whose `tag` matches is aborted by shutting its socket down.
1469. /// The peer sees the disconnect (llama-server / OpenRouter stop generating and
1470. /// billing), the worker's read unblocks, and the FINAL completion frame arrives
1471. /// with `aborted = true` and error "hl:fetch: aborted". Returns how many were
1472. /// aborted (0 = nothing in flight under that tag — already finished, or a typo).
1473. export fn hl_fetch_abort(argc: u32, argv: [*]const HlValue) callconv(.c) HlValue {
1474. if (argc < 1 or argv[0].type != .hl_string) {
1475. return api.makeError("hl:fetch abort: pass the tag the fetch was started with");
1476. }
1477. const tag = argv[0].data.string.ptr[0..argv[0].data.string.len];
1478. var n: u32 = 0;
1479. mutexLock(&active_mutex);
1480. for (active_tickets.items) |t| {
1481. if (std.mem.eql(u8, t.req.tag, tag)) {
1482. t.abort();
1483. n += 1;
1484. }
1485. }
1486. mutexUnlock(&active_mutex);
1487. return api.makeNumber(@floatFromInt(n));
1488. }
1490. /// __native("fetch.completions") → the loop source. Calling it ARMS completion
1491. /// delivery for every fetch started from here on.
1492. export fn hl_fetch_completions(argc: u32, argv: [*]const HlValue) callconv(.c) HlValue {
1493. _ = argc;
1494. _ = argv;
1495. mutexLock(&comp_mutex);
1496. if (comp_wake_fd < 0) comp_wake_fd = http.makeWakeFd();
1497. mutexUnlock(&comp_mutex);
1498. comp_enabled.store(true, .release);
1500. const iter = allocator.create(HlIterator) catch return api.makeError("hl:fetch: out of memory");
1501. iter.* = .{
1502. .context = null,
1503. .next_fn = &completionsTryNext, // non-blocking either way — event loop only
1504. .deinit_fn = null,
1505. .try_next_fn = &completionsTryNext,
1506. .wake_fd = comp_wake_fd,
1507. };
1508. return api.makeIterator(iter);
1509. }